""" 棉花糖,序列化工具 核心方法: 序列化:schema.dump(object)、schema.dumps(object) 反序列化:schema.load(dict) """ import uuid from datetime import datetime from pprint import pprint from marshmallow import Schema, fields, post_load, ValidationError, validate """ 序列化和反序列化 """ print(" v v v") print("创建自定义对象----------------------------------------v v") print(" v") # 创建自定义对象object class User: def __init__(self, name, email, created_at): self.name = name self.email = email self.created_at = created_at print(" v v v") print("创建Schema的两种方式----------------------------------v v") print(" v") # 创建结构描述 # 方式一 class UserSchema(Schema): name = fields.Str() email = fields.Email() created_at = fields.DateTime() @post_load def make_user(self, data, **kwargs): return User(**data) # 方式二 UserSchema2 = Schema.from_dict({"name": fields.Str(), "email": fields.Email(), "created_at": fields.DateTime()}) print(" v v v") print("序列化对象示例----------------------------------------v v") print(" v") # 序列化对象 user = User("zht", "2862058843@163.com", datetime.now()) userSchema = UserSchema() result = userSchema.dump(user) # 美化输出 pprint(result) print(type(result)) userSchema2 = UserSchema2() result2 = userSchema2.dump(user) pprint(result2) # print(result) # 转成json字符串 json_result = userSchema2.dumps(user) pprint(json_result) print(type(json_result)) print(" v v v") print("序列化对象示例(过滤输出)-------------------------------v v") print(" v") # 过滤输出 userSchema2 = UserSchema2(only=('name', 'email')) json_result = userSchema2.dump(user) pprint(json_result) print(type(json_result)) print(json_result['name']) print(" v v v") print("返序列化对象示例--------------------------------------v v") print(" v") # 反序列化对象 注意的是 字典里的key,对象构造函数一定要有此key user_data = { "created_at": "2014-08-11T05:26:03.869245", "email": "ken@yahoo.com", "name": "Ken", } userSchema3 = UserSchema() result = userSchema3.load(user_data) pprint(result) print(type(result)) # 反序列化对象集合 user1 = User("zht1", "2862058843@163.com", datetime.now()) user2 = User("zht2", "2862058843@164.com", datetime.now()) user3 = User("zht3", "2862058843@165.com", datetime.now()) users = [user1, user2, user3] userSchemas = UserSchema(many=True) result = userSchemas.dump(users) pprint(result) """ 验证属性值 """ print(" v v v") print("验证属性值示例----------------------------------------v v") print(" v") # 验证一个对象 print("验证一个对象") try: result = UserSchema().load({"name": "zht", "email": "zht"}) print(result) except ValidationError as e: print(e.messages) print(e.valid_data) # 验证多个对象 print("验证多个对象") class BandMemberSchema(Schema): name = fields.String(required=True) email = fields.Email() user_data = [ {"email": "mick@stones.com", "name": "Mick"}, {"email": "invalid", "name": "Invalid"}, # invalid email {"email": "keith@stones.com", "name": "Keith"}, {"email": "charlie@stones.com"}, # missing "name" ] try: BandMemberSchema(many=True).load(user_data) print(result) except ValidationError as e: print(e.messages) print(e.valid_data) # 通过传递 validate参数传递验证器对字段进行验证 print("通过传递 validate参数传递验证器对字段进行验证") class MemberSchema(Schema): name = fields.Str(validate=validate.Length(min=1)) # 最小字符长度 permission = fields.Str(validate=validate.OneOf(["read", "write", "admin"])) # 必须是其中一个 age = fields.Int(validate=validate.Range(min=18, max=40)) # 年龄必须是18到40 in_data = {"name": "1", "permission": "invalid", "age": 71} try: MemberSchema().load(in_data) except ValidationError as e: print(e.messages) # 输出针对每个字段的错误信息 print(e.valid_data) # 输出有效的字段 (验证通过的字段) # 自己实现一个验证器 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": -1} try: result = ItemSchema().load(in_data) print(result) except ValidationError as e: print(e.messages) # 输出针对每个字段的错误信息 print(e.valid_data) # 输出有效的字段 (验证通过的字段) # 自定义错误信息 print("自定义字段验证required的错误消息") class TRequiredSchema(Schema): name = fields.String(required=True) age = fields.Integer(required=True, error_messages={"required": "Age is required."}) city = fields.String( required=True, error_messages={"required": {"message": "City required", "code": 400}}, ) email = fields.Email() try: result = TRequiredSchema().load({"email": "foo@bar.com"}) except ValidationError as err: pprint(err.messages) # {'age': ['Age is required.'], # 'city': {'code': 400, 'message': 'City required'}, # 'name': ['Missing data for required field.']} print("指定默认值") # 指定默认 class DefaultSchema(Schema): id = fields.UUID(missing=uuid.uuid1) birthdate = fields.DateTime(default=datetime(2017, 9, 29)) name = fields.String(default="火") print(DefaultSchema().load({})) # {'id': UUID('337d946c-32cd-11e8-b475-0022192ed31b')} print(DefaultSchema().dump({})) # {'birthdate': '2017-09-29T00:00:00+00:00'}