在 api 请求的的参数校验中,使用 marshmallow 非常方便
如果要针对大量json数据做检验,推荐使用 python-jsonschema
1. 定义Schema
# 定义模型
import datetime as dt
class User:
def __init__(self, name, email):
self.name = name
self.email = email
self.created_at = dt.datetime.now()
def __repr__(self):
return "".format(self=self)
# 定义Schema
class UserSchema(Schema):
name = fields.String()
email = fields.Email()
create_at = fields.DateTime()
2. 使用动态Schema
from marshmallow import Schema, fields
UserSchema = Schema.from_dict(
{"name": fields.Str(), "email": fields.Email(), "created_at": fields.DateTime()}
)
3. dump(obj -> dict)
from pprint import pprint
user = User(name="Monty", email="monty@python.org")
schema = UserSchema()
result = schema.dump(user)
pprint(result)
#{'create_at': '2024-06-16T21:37:04.711994',
# 'email': 'monty@python.org',
# 'name': 'Monty'}
print(result["name"]) # Monty
4. dumps(obj -> str)
json_result = schema.dumps(user)
pprint(json_result)
# '{"name": "Monty", "email": "monty@python.org", "created_at": "2014-08-17T14:54:16.049594+00:00"}'*
输出过滤
only
summary_schema = UserSchema(only=("name", "email"))
summary_schema.dump(user)
# {"name": "Monty", "email": "monty@python.org"}
exculde
summary_schema = UserSchema(exclude=("email",))
s = summary_schema.dump(user)
print(s)
# {'name': 'Monty', 'create_at': '2024-06-16T21:43:28.035327'}
5. load
Deserializing - > dict (校验)
fields.DateTime()
对象会被序列化为python日期对象
user_data = {
"created_at": "2014-08-11T05:26:03.869245",
"email": "ken@yahoo.com",
"name": "Ken",
}
schema = UserSchema()
result = schema.load(user_data)
pprint(result)
#{'created_at': datetime.datetime(2014, 8, 11, 5, 26, 3, 869245),
# 'email': 'ken@yahoo.com',
# 'name': 'Ken'}
print(type(result))
Deserializing -> obj
为了将数据反序列化为对象,请为模式定义一个方法,并用 post_load 对其进行装饰。该方法接收反序列化数据的字典。
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, **kwargs):
return User(**data)
使用load方法返回一个User实例
user_data = {
"created_at": "2014-08-11T05:26:03.869245",
"email": "ken@yahoo.com",
"name": "Ken",
}
schema = UserSchema()
result = schema.load(user_data)
pprint(result) #
print(result.email) # ken@yahoo.com
6. 处理集合对象
当处理可迭代的集合对象时,设置`many=True``
dump
user1 = User(name="Mick", email="mick@tones.com")
user2 = User(name="Keith", email="keith@stones.com")
users = [user1, user2]
schema = UserSchema(many=True)
result = schema.dump(users) # OR UserSchema().dump(users, many=True)
pprint(result)
#[{'created_at': '2024-06-18T21:29:14.942278',
# 'email': 'mick@tones.com',
# 'name': 'Mick'},
# {'created_at': '2024-06-18T21:29:14.942287',
# 'email': 'keith@stones.com',
# 'name': 'Keith'}]
load
s = [{'created_at': '2024-06-18T21:29:14.942278',
'email': 'mick@tones.com',
'name': 'Mick'},
{'created_at': '2024-06-18T21:29:14.942287',
'email': 'keith@stones.com',
'name': 'Keith'}]
schema = UserSchema(many=True)
result = schema.load(s) # UserSchema().load(s, many=True)
print(result)
7. Validation校验
单个校验
try:
result = UserSchema().load({"name": "John", "email": "foo"})
except ValidationError as err:
print(err.messages) # => {"email": ['"foo" is not a valid email address.']}
print(err.valid_data) # => {"name": "John"}
校验集合
from marshmallow import Schema, fields, post_load, ValidationError
class UserSchema(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:
UserSchema(many=True).load(user_data)
except ValidationError as err:
pprint(err.messages)
# {1: {'email': ['Not a valid email address.']},
# 3: {'name': ['Missing data for required field.']}}
8. 添加校验参数
from pprint import pprint
from marshmallow import Schema, fields, validate, ValidationError
class UserSchema(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))
in_data = {"name": "", "permission": "invalid", "age": 71}
try:
UserSchema().load(in_data)
except ValidationError as err:
pprint(err.messages)
# {'age': ['Must be greater than or equal to 18 and less than or equal to 40.'],
# 'name': ['Shorter than minimum length 1.'],
# 'permission': ['Must be one of: read, write, admin.']}
9. 自定义校验规则
校验函数接收一个单一的参数,校验失败应该raise一个ValidationError或者返回False
from marshmallow import Schema, fields, ValidationError
def validate_quantity(n):
if n < 0:
raise ValidationError("Quantity must be greater than 0.")
if 30 < n <= 60:
raise ValidationError("Quantity must not be greater than 30.")
if n > 60:
return False
class ItemSchema(Schema):
quantity = fields.Integer(validate=validate_quantity)
in_data = {"quantity": 50}
try:
result = ItemSchema().load(in_data)
except ValidationError as err:
print(err.messages)
也可以给定一个校验函数的集合(list, tuple, generator)
def validate_quantity_2(n):
if n%2 ==0:
raise ValidationError("The quantity must not be even number")
class ItemSchema(Schema):
quantity = fields.Integer(validate=[validate_quantity, validate_quantity_2])
# {'quantity': ['Quantity must not be greater than 30.', 'The quantity must not be even number']}
注意
- 验证在反序列化时进行,但不在序列化时进行。为提高序列化性能,传递给 Schema.dump() 的数据被视为有效数据。
...