摘要：創(chuàng)建實(shí)例時(shí)如果傳遞了，表示需要接收輸入數(shù)據(jù)集合，裝飾器注冊(cè)預(yù)處理和后處理方法時(shí)需要傳遞參數(shù)。

數(shù)據(jù)的預(yù)處理和后處理方法通過(guò)pre_load, post_load, pre_dump和post_dump裝飾器注冊(cè)：

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"

預(yù)處理和后處理方法默認(rèn)一次接收一個(gè)對(duì)象/數(shù)據(jù)，在運(yùn)行時(shí)處理傳遞給schema對(duì)象的many參數(shù)。

創(chuàng)建schema實(shí)例時(shí)如果傳遞了many=True，表示需要接收輸入數(shù)據(jù)集合，裝飾器注冊(cè)預(yù)處理和后處理方法時(shí)需要傳遞參數(shù)pass_many=True。預(yù)處理和后處理方法接收輸入數(shù)據(jù)(可能是單個(gè)數(shù)據(jù)或數(shù)據(jù)集合)和布爾類型的many參數(shù)：

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="[email protected]")
user_data = user_schema.dump(user).data
# {"user": {"email": "[email protected]", "name": "Mick"}}

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

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

字段驗(yàn)證產(chǎn)生的錯(cuò)誤字典的_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."]}

如果不想存儲(chǔ)在_schema鍵中，可以指定新的鍵名傳遞給ValidationError的第二個(gè)參數(shù)：

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的區(qū)別)：

@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參數(shù)裝飾的方法的調(diào)用順序

重寫(xiě)schema的handle_error方法來(lái)自定義錯(cuò)誤處理功能。handle_error接收一個(gè)ValidationError異常實(shí)例，一個(gè)原始對(duì)象(序列化)或輸入數(shù)據(jù)(反序列化)：

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注冊(cè)一個(gè)schema級(jí)別的驗(yàn)證函數(shù)，其異常信息保存在錯(cuò)誤字典的_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"]

通常驗(yàn)證器會(huì)忽略未聲明的field的數(shù)據(jù)輸入。如果要訪問(wèn)原始輸入數(shù)據(jù)(例如如果發(fā)送了未知字段視為驗(yàn)證失敗)，可以給validates_schema裝飾器傳遞一個(gè)pass_original=True參數(shù)：

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級(jí)別的驗(yàn)證錯(cuò)誤，可以給ValidationError的第二個(gè)參數(shù)傳遞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默認(rèn)使用utils.get_value函數(shù)獲取各種類型的對(duì)象的屬性以進(jìn)行序列化。

通過(guò)重寫(xiě)get_attribute方法可以重寫(xiě)對(duì)象屬性的訪問(wèn)方式：

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行為的一種方式。通過(guò)繼承自SchemaOpts可以添加自定義class Meta選項(xiàng)(Schema.Meta API docs查看原生選項(xiàng))。

下面的代碼通過(guò)自定義class Meta選項(xiàng)實(shí)現(xiàn)了預(yù)處理和后處理的many參數(shù)這一節(jié)中例子的功能。

首先通過(guò)繼承SchemaOpts類添加了兩個(gè)選項(xiàng)，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)

然后創(chuàng)建NamespacedSchema類并使用剛才創(chuàng)建的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}

現(xiàn)在我們處理序列化和反序列化的自定義schema再繼承自NamespacedSchema：

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

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

ser = UserSchema()
user = User("Keith", email="[email protected]")
result = ser.dump(user)
result.data  # {"user": {"name": "Keith", "email": "[email protected]"}}

Schema的context屬性存儲(chǔ)序列化及反序列化可能要用到的額外信息。

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

我的博客即將同步至騰訊云+社區(qū)，邀請(qǐng)大家一同入駐：https://cloud.tencent.com/dev...