Python（二）Marshmallow 库相关学习

西凉风雷

发布于 2022-11-23 19:32:49

45300

代码可运行

文章被收录于专栏：容器与 Kubernetes 游记容器与 Kubernetes 游记

运行总次数：0

代码可运行

0. 前言

Marshmallow 是一个用于将 ORM 对象与 Python 原生数据类型之间转换的库。实现 object → dict、object → list、string → dict 和 string → list 等功能

1. Schema

实现一个 object 和 json 之间的转化需要一个 Schema 对象作为中间载体，同时实现校验的功能：

class ImageTpl(Schema):
    value = fields.Dict(required=True, error_messages=get_field_valid_msg("图片模版值"))
    height = fields.Int(required=True, error_messages=get_field_valid_msg("图片模版高度"))
    width = fields.Int(required=True, error_messages=get_field_valid_msg("图片模版宽度"))

2. 序列化

序列化使用 dump() 或者 dumps() 方法，其中 dump() 实现 object → dict，dumps() 实现 object → string：

from marshmallow import pprint

user = User(name="Monty", email="monty@python.org")
schema = UserSchema()
result = schema.dump(user)
pprint(result.data)
# {"name": "Monty",
#  "email": "monty@python.org",
#  "created_at": "2014-08-17T14:54:16.049594+00:00"}

序列化的结果可以通过 only 参数指定字段：

from pprint import pprint

user_data = {
    'created_at': '2014-08-11T05:26:03.869245',
    'email': u'ken@yahoo.com',
    'name': u'Ken'
}
schema = UserSchema()
result = schema.load(user_data)
pprint(result.data)
# {'name': 'Ken',
#  'email': 'ken@yahoo.com',
#  'created_at': datetime.datetime(2014, 8, 11, 5, 26, 3, 869245)},

3. 反序列化

反序列化使用 load() 或者 loads() 方法，分别实现 dict → object 和 string → object。其中 dict → object 需要添加装饰器，自己实现逻辑：

from marshmallow import Schema, fields, post_load

class UserSchema(Schema):
    name = fields.Str()
    email = fields.Email()
    created_at = fields.DateTime()

    @post_load
    def make_user(self, data):
        return User(**data)

user_data = {
    'name': 'Ronnie',
    'email': 'ronnie@stones.com'
}
schema = UserSchema()
result = schema.load(user_data)
result.data  # => <User(name='Ronnie')>

4. Field 对象

Schema 对象为每个属性赋值为一个 Field 对象设定转换类型的校验参数，具体如下：
validate 参数：指定一个 lambda 函数或者函数，定义校验逻辑，传入函数定义了 ValidationError 的话，返回信息会记录抛出的异常：

from marshmallow import Schema, fields, ValidationError

def validate_quantity(n):
    if n < 0:
        raise ValidationError('Quantity must be greater than 0.')
    if n > 30:
        raise ValidationError('Quantity must not be greater than 30.')

class ItemSchema(Schema):
    quantity = fields.Integer(validate=validate_quantity)

in_data = {'quantity': 31}
result, errors = ItemSchema().load(in_data)
errors  # => {'quantity': ['Quantity must not be greater than 30.']}

required 参数：标记该字段必须传递切被校验
error_messages 参数：传递字典定义错误返回信息：

def get_field_valid_msg(field_name):
    return {
        'required': '[%s] 字段必填' % field_name,
        'type': '[%s] 字段类型不合法' % field_name, # used by Unmarshaller
        'null': '[%s] 字段不能为空' % field_name,
        'validator_failed': '[%s] 字段值不合法' % field_name
    }

many 参数：同 fields.Nested 同用表示指定元素类型的数组类型，严格按照数据类型校验，fields.Nested 表示制定另一个 Schema 作为外键：

......
'components': fields.Nested(Component, required=True, many=True,
                                    error_messages=get_field_valid_msg("组件信息")),
......

fields.List 类型：一类 Field 元素，可以指定元素类型，对于传递的非数组类型的元素，会自动包装为一个数组

5. 验证

对于 Schema 的校验有很多方式，通过上述 Field 元素的相关参数已经实现了很多对元素的校验
也可以通过定义 @validates(field_name) 装饰器定义特定属性的校验函数
也可以通过定义 @validate_schema() 装饰器定义 Schema 级别的校验函数：

@validates_schema
    def validate_element_type(self, value):

        switch = {
            DpaVideoPackageElementType.TEXT : lambda x : x in DpaVideoPackageDataType.text_types(),
            DpaVideoPackageElementType.IMAGE : lambda x : x in DpaVideoPackageDataType.image_types(),
            DpaVideoPackageElementType.LOGO : lambda x : x in DpaVideoPackageDataType.logo_types()
        }

        try:
            if not switch[value['element_type']](value['data_type']):
                raise ValidationError("组件类型和组件数据类型不匹配")
        except KeyError as e:
            raise ValidationError("组件类型不存在")

对于一个 Schema，load() 和 loads 方法会在返回值中加入验证错误的信息：

class ImageOrImageMeta(Schema):
    value = fields.String(required=True, error_messages=get_field_valid_msg("图片或图片元数据值"))
    height = fields.Int(required=True, error_messages=get_field_valid_msg("图片或图片元数据高度"))
    width = fields.Int(required=True, error_messages=get_field_valid_msg("图片或图片元数据宽度"))

class ImageTpl(Schema):
    value = fields.Dict(required=True, error_messages=get_field_valid_msg("图片模版值"))
    height = fields.Int(required=True, error_messages=get_field_valid_msg("图片模版高度"))
    width = fields.Int(required=True, error_messages=get_field_valid_msg("图片模版宽度"))

class TextOrTextMeta(Schema):
    value = fields.String(required=True, error_messages=get_field_valid_msg("文本或文本元数据值"))
    default_val = fields.String(error_messages=get_field_valid_msg("文本或文本元数据默认值"))
    max_length = fields.Int(required=True, error_messages=get_field_valid_msg("文本或文本元数据最大值"))
......
        switch = {
            DpaVideoPackageDataType.IMAGE : lambda x : ImageOrImageMeta().load(x),
            DpaVideoPackageDataType.IMAGE_META : lambda x : ImageOrImageMeta().load(x),
            DpaVideoPackageDataType.IMAGE_TPL : lambda  x : ImageTpl().load(x),
            DpaVideoPackageDataType.TEXT : lambda  x : TextOrTextMeta().load(x),
            DpaVideoPackageDataType.TEXT_META : lambda  x : TextOrTextMeta().load(x),
        }

        result = switch[value['data_type']](value['data'])
        if result.errors:
            raise ValidationError(result.errors)