marshmallow之Schema延伸功能

摘要：创建实例时如果传递了，表示需要接收输入数据集合，装饰器注册预处理和后处理方法时需要传递参数。

数据的预处理和后处理方法通过pre_load, post_load, pre_dump和post_dump装饰器注册：

from marshmallow import Schema, fields, pre_load

class UserSchema(Schema):
    name = fields.Str()
    slug = fields.Str()

    @pre_load
    def slugify_name(self, in_data):
        in_data["slug"] = in_data["slug"].lower().strip().replace(" ", "-")
        return in_data

schema = UserSchema()
result, errors = schema.load({"name": "Steve", "slug": "Steve Loria "})
result["slug"]  # => "steve-loria"

预处理和后处理方法默认一次接收一个对象/数据，在运行时处理传递给schema对象的many参数。

创建schema实例时如果传递了many=True，表示需要接收输入数据集合，装饰器注册预处理和后处理方法时需要传递参数pass_many=True。预处理和后处理方法接收输入数据(可能是单个数据或数据集合)和布尔类型的many参数：

from marshmallow import Schema, fields, pre_load, post_load, post_dump

class BaseSchema(Schema):
    # Custom options
    __envelope__ = {
        "single": None,
        "many": None
    }
    __model__ = User

    def get_envelope_key(self, many):
        """Helper to get the envelope key."""
        key = self.__envelope__["many"] if many else self.__envelope__["single"]
        assert key is not None, "Envelope key undefined"
        return key

    @pre_load(pass_many=True)
    def unwrap_envelope(self, data, many):
        key = self.get_envelope_key(many)
        return data[key]

    @post_dump(pass_many=True)
    def wrap_with_envelope(self, data, many):
        key = self.get_envelope_key(many)
        return {key: data}

    @post_load
    def make_object(self, data):
        return self.__model__(**data)

class UserSchema(BaseSchema):
    __envelope__ = {
        "single": "user",
        "many": "users",
    }
    __model__ = User
    name = fields.Str()
    email = fields.Email()

user_schema = UserSchema()

user = User("Mick", email="mick@stones.org")
user_data = user_schema.dump(user).data
# {"user": {"email": "mick@stones.org", "name": "Mick"}}

users = [User("Keith", email="keith@stones.org"),
        User("Charlie", email="charlie@stones.org")]
users_data = user_schema.dump(users, many=True).data
# {"users": [{"email": "keith@stones.org", "name": "Keith"},
#            {"email": "charlie@stones.org", "name": "Charlie"}]}

user_objs = user_schema.load(users_data, many=True).data
# [, ]

字段验证产生的错误字典的_schema键包含了ValidationError异常的信息：

from marshmallow import Schema, fields, ValidationError, pre_load

class BandSchema(Schema):
    name = fields.Str()

    @pre_load
    def unwrap_envelope(self, data):
        if "data" not in data:
            raise ValidationError("Input data must have a "data" key.")
        return data["data"]

sch = BandSchema()
sch.load({"name": "The Band"}).errors
# {"_schema": ["Input data must have a "data" key."]}

如果不想存储在_schema键中，可以指定新的键名传递给ValidationError的第二个参数：

from marshmallow import Schema, fields, ValidationError, pre_load

class BandSchema(Schema):
    name = fields.Str()

    @pre_load
    def unwrap_envelope(self, data):
        if "data" not in data:
            raise ValidationError("Input data must have a "data" key.", "_preprocessing")
        return data["data"]

sch = BandSchema()
sch.load({"name": "The Band"}).errors
# {"_preprocessing": ["Input data must have a "data" key."]}

反序列化的处理流程：

@pre_load(pass_many=True) methods

@pre_load(pass_many=False) methods

load(in_data, many) (validation and deserialization)

@post_load(pass_many=True) methods

@post_load(pass_many=False) methods

序列化的处理流程(注意pass_many的区别)：

@pre_dump(pass_many=False) methods

@pre_dump(pass_many=True) methods

dump(obj, many) (serialization)

@post_dump(pass_many=False) methods

@post_dump(pass_many=True) methods

不保证相同装饰器和pass_many参数装饰的方法的调用顺序

重写schema的handle_error方法来自定义错误处理功能。handle_error接收一个ValidationError异常实例，一个原始对象(序列化)或输入数据(反序列化)：

import logging
from marshmallow import Schema, fields

class AppError(Exception):
    pass

class UserSchema(Schema):
    email = fields.Email()

    def handle_error(self, exc, data):
        """Log and raise our custom exception when (de)serialization fails."""
        logging.error(exc.messages)
        raise AppError("An error occurred with input: {0}".format(data))

schema = UserSchema()
schema.load({"email": "invalid-email"})  # raises AppError

使用marshmallow.validates_schema装饰器可以为Schema注册一个schema级别的验证函数，其异常信息保存在错误字典的_schema键中：

from marshmallow import Schema, fields, validates_schema, ValidationError

class NumberSchema(Schema):
    field_a = fields.Integer()
    field_b = fields.Integer()

    @validates_schema
    def validate_numbers(self, data):
        if data["field_b"] >= data["field_a"]:
            raise ValidationError("field_a must be greater than field_b")

schema = NumberSchema()
result, errors = schema.load({"field_a": 1, "field_b": 2})
errors["_schema"] # => ["field_a must be greater than field_b"]

通常验证器会忽略未声明的field的数据输入。如果要访问原始输入数据(例如如果发送了未知字段视为验证失败)，可以给validates_schema装饰器传递一个pass_original=True参数：

from marshmallow import Schema, fields, validates_schema, ValidationError

class MySchema(Schema):
    foo = fields.Int()
    bar = fields.Int()

    @validates_schema(pass_original=True)
    def check_unknown_fields(self, data, original_data):
        unknown = set(original_data) - set(self.fields)
        if unknown:
            raise ValidationError("Unknown field", unknown)

schema = MySchema()
errors = schema.load({"foo": 1, "bar": 2, "baz": 3, "bu": 4}).errors
# {"baz": "Unknown field", "bu": "Unknown field"}

如果要在指定field上保存schema级别的验证错误，可以给ValidationError的第二个参数传递field名称(列表)：

class NumberSchema(Schema):
    field_a = fields.Integer()
    field_b = fields.Integer()

    @validates_schema
    def validate_numbers(self, data):
        if data["field_b"] >= data["field_a"]:
            raise ValidationError(
                "field_a must be greater than field_b",
                "field_a"
            )

schema = NumberSchema()
result, errors = schema.load({"field_a": 1, "field_b": 2})
errors["field_a"] # => ["field_a must be greater than field_b"]

marshmallow默认使用utils.get_value函数获取各种类型的对象的属性以进行序列化。

通过重写get_attribute方法可以重写对象属性的访问方式：

class UserDictSchema(Schema):
    name = fields.Str()
    email = fields.Email()

    # If we know we"re only serializing dictionaries, we can
    # use dict.get for all input objects
    def get_attribute(self, key, obj, default):
        return obj.get(key, default)

class Meta是配置和修改Schema行为的一种方式。通过继承自SchemaOpts可以添加自定义class Meta选项(Schema.Meta API docs查看原生选项)。

下面的代码通过自定义class Meta选项实现了预处理和后处理的many参数这一节中例子的功能。

首先通过继承SchemaOpts类添加了两个选项，name和plural_name：

from marshmallow import Schema, SchemaOpts

class NamespaceOpts(SchemaOpts):
    """Same as the default class Meta options, but adds "name" and
    "plural_name" options for enveloping.
    """
    def __init__(self, meta):
        SchemaOpts.__init__(self, meta)
        self.name = getattr(meta, "name", None)
        self.plural_name = getattr(meta, "plural_name", self.name)

然后创建NamespacedSchema类并使用刚才创建的NamespaceOpts：

class NamespacedSchema(Schema):
    OPTIONS_CLASS = NamespaceOpts

    @pre_load(pass_many=True)
    def unwrap_envelope(self, data, many):
        key = self.opts.plural_name if many else self.opts.name
        return data[key]

    @post_dump(pass_many=True)
    def wrap_with_envelope(self, data, many):
        key = self.opts.plural_name if many else self.opts.name
        return {key: data}

现在我们处理序列化和反序列化的自定义schema再继承自NamespacedSchema：

class UserSchema(NamespacedSchema):
    name = fields.String()
    email = fields.Email()

    class Meta:
        name = "user"
        plural_name = "users"

ser = UserSchema()
user = User("Keith", email="keith@stones.com")
result = ser.dump(user)
result.data  # {"user": {"name": "Keith", "email": "keith@stones.com"}}

Schema的context属性存储序列化及反序列化可能要用到的额外信息。

schema = UserSchema()
# Make current HTTP request available to
# custom fields, schema methods, schema validators, etc.
schema.context["request"] = request
schema.dump(user)

我的博客即将同步至腾讯云+社区，邀请大家一同入驻：https://cloud.tencent.com/dev...