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'

預處理和後處理的many參數

預處理和後處理方法默認一次接收一個對象/數據，在運行時處理傳遞給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
# [<User(name='Keith Richards')>, <User(name='Charlie Watts')>]

在預處理和後處理方法中拋出異常

字段驗證產生的錯誤字典的_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.']}

預處理和後處理方法的調用順序

反序列化的處理流程：

1. @pre_load(pass_many=True) methods
2. @pre_load(pass_many=False) methods
3. load(in_data, many) (validation and deserialization)
4. @post_load(pass_many=True) methods
5. @post_load(pass_many=False) methods

序列化的處理流程(注意pass_many的區別)：

1. @pre_dump(pass_many=False) methods
2. @pre_dump(pass_many=True) methods
3. dump(obj, many) (serialization)
4. @post_dump(pass_many=False) methods
5. @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

Schema級別的驗證

使用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的錯誤

若是要在指定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選項

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...