浅析Marshmallow在flask中的应用

洪季萌

2023-12-01

Marshmallow

官方文档：https://marshmallow.readthedocs.io/en/latest/

Marshmallow，中文译作：棉花糖。是一个轻量级的数据格式转换的模块，也叫序列化和反序列化模块，常用于将复杂的orm模型对象与python原生数据类型之间相互转换。marshmallow提供了丰富的api功能。如下：

Serializing

序列化[可以把数据对象转化为可存储或可传输的数据类型，例如：objects/object->list/dict，dict/list->string]

Deserializing

反序列化器[把可存储或可传输的数据类型转换成数据对象，例如：list/dict->objects/object，string->dict/list]

Validation

数据校验，可以在反序列化阶段，针对要转换数据的内容进行类型验证或自定义验证。

Marshmallow本身是一个单独的库，基于我们当前项目使用框架是flask并且数据库ORM框架使用SQLAlchemy，所以我们可以通过安装flask-sqlalchemy和marshmallow-sqlalchemy集成到项目就可以了。

基本安装和配置

模块安装：

pip install -U marshmallow-sqlalchemy
pip install -U flask-sqlalchemy
pip install -U flask-marshmallow

模块初始化：

import os

from flask import Flask
from flask_script import Manager
from flask_sqlalchemy import SQLAlchemy
from flask_redis import FlaskRedis
from flask_session import Session
from flask_migrate import Migrate,MigrateCommand
from flask_jsonrpc import JSONRPC
from flask_marshmallow import Marshmallow

from application.utils import init_blueprint
from application.utils.config import load_config
from application.utils.session import init_session
from application.utils.logger import Log
from application.utils.commands import load_command
# 创建终端脚本管理对象
manager = Manager()

# 创建数据库链接对象
db = SQLAlchemy()

# redis链接对象
redis = FlaskRedis()

# Session存储对象
session_store = Session()

# 数据迁移实例对象
migrate = Migrate()

# 日志对象
log = Log()

# jsonrpc模块实例对象
jsonrpc = JSONRPC()

# 数据转换器的对象创建
ma = Marshmallow()

def init_app(config_path):
    """全局初始化"""
    # 创建app应用对象
    app = Flask(__name__)
    # 项目根目录
    app.BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

    # 加载配置
    Config = load_config(config_path)
    app.config.from_object(Config)

    # 数据库初始化
    db.init_app(app)
    redis.init_app(app)

    # 数据转换器的初始化
    ma.init_app(app)

    # session存储初始化
    init_session(app)
    session_store.init_app(app)

    # 数据迁移初始化
    migrate.init_app(app,db)
    # 添加数据迁移的命令到终端脚本工具中
    manager.add_command('db', MigrateCommand)

    # 日志初始化
    app.log = log.init_app(app)

    # 蓝图注册
    init_blueprint(app)

    # jsonrpc初始化
    jsonrpc.service_url = "/api" # api接口的url地址前缀
    jsonrpc.init_app(app)

    # 初始化终端脚本工具
    manager.app = app

    # 注册自定义命令
    load_command(manager)

    return manager

为了方便学习和使用Marshllow, 我们单独创建一个marsh蓝图来验证这个模块的基本使用.

cd application/apps
python ../../manage.py blue -n marsh

注册marsh模块,application.settings.dev代码:

    INSTALLED_APPS = [
        "application.apps.home",
        "application.apps.users",
        "application.apps.marsh",
    ]

总路由,application.urls 代码:

from application.utils import include
urlpatterns = [
    include("","home.urls"),
    include("/users","users.urls"),
    include("/marsh","marsh.urls"),
]

视图application.apps.marsh.views代码:

from marshmallow import Schema,fields
from application.apps.users.models import User,UserProfile
class UserSchema(Schema):
    name   = fields.String()
    age    = fields.Integer()
    email  = fields.Email()
    money  = fields.Number()
    class Meta:
        fields = ["name","age","money","email","info"]
        ordered = True # 转换成有序字典

def index():
    """序列化"""
    return "ok" # flask==1.1.1以后,可以直接返回字典数据,flask1.0以前必须通过jsonify返回字典

子路由,application.apps.marsh.urls,代码:

from . import views
from application.utils import path
urlpatterns = [
    path("", views.index),
]

基本构造器(Schema)

marshmallow转换数据格式主要通过构造器类来完成，而Schema类提供了数据转换的基本功能：序列化，验证和反序列化。所以在使用marshmallow的过程中所有的构造器类必须直接或间接继承于Schema基类

基于Schema完成数据序列化转换

application.apps.marsh.views代码:

from marshmallow import Schema,fields
from application.apps.users.models import User,UserProfile
class UserSchema(Schema):
    name   = fields.String()
    age    = fields.Integer()
    email  = fields.Email()
    money  = fields.Number()
    class Meta:
        fields = ["name","age","money","email","info"]
        ordered = True # 转换成有序字典

def index():
    """序列化"""
    """单个模型数据的序列化处理"""
    user1 = User( name="xiaoming", password="123456", age=16, email="333@qq.com", money=31.50 )
    # print(user1)
    # 把模型对象转换成字典格式
    data1 = UserSchema().dump(user1)
    print(type(data1),data1)
    
    # 把模型对象转换成json字符串格式
    data2 = UserSchema().dumps(user1)
    print(type(data2), data2)
    return "ok"

schema常用属性数据类型

类型	描述
fields.`Dict`(keys, type]] = None, values, …)	字典类型，常用于接收json类型数据
fields.`List`(cls_or_instance, type], **kwargs)	列表类型，常用于接收数组数据
fields.`Tuple`(tuple_fields, args, *kwargs)	元组类型
fields.`String`(*, default, missing, data_key, …)	字符串类型
fields.`UUID`(*, default, missing, data_key, …)	UUID格式类型的字符串
fields.`Number`(, as_string, *kwargs)	数值基本类型
fields.`Integer`(, strict, *kwargs)	整型
fields.`Decimal`(places, rounding, *, allow_nan, …)	数值型
fields.`Boolean`(, truthy, falsy, *kwargs)	布尔型
fields.`Float`(, allow_nan, as_string, *kwargs)	浮点数类型
fields.`DateTime`(format, **kwargs)	日期时间类型
fields.`Time`(format, **kwargs)	时间类型
fields.`Date`(format, **kwargs)	日期类型
fields.`Url`(*, relative, schemes, Set[str]]] = None, …)	url网址字符串类型
fields.`Email`(args, *kwargs)	邮箱字符串类型
fields.`IP`(*args[, exploded])	IP地址字符串类型
fields.`IPv4`(*args[, exploded])	IPv4地址字符串类型
fields.`IPv6`(*args[, exploded])	IPv6地址字符串类型
fields.`Method`(serialize, deserialize, **kwargs)	基于Schema类方法返回值的字段
fields.`Function`(serialize, Any], Callable[[Any, …)	基于函数返回值得字段
fields.`Nested`(nested, type, str, Callable[[], …)	外键类型

Schema数据类型的常用通用属性

属性名	描述
default	序列化阶段中设置字段的默认值
missing	反序列化阶段中设置字段的默认值
validate	反序列化阶段调用的内置数据验证器或者内置验证集合
required	设置当前字段的必填字段
allow_none	是否允许为空
load_only	是否在反序列化阶段才使用到当前字段
dump_omly	是否在序列化阶段才使用到当前字段
error_messages	字典类型，可以用来替代默认的字段异常提示语，格式： error_messages={“required”: “用户名为必填项。”}

在前面进行的序列化操作属于序列化单个数据对象, MarshMallow中也可以进行多个数据对象的序列化.

from marshmallow import Schema,fields
from application.apps.users.models import User,UserProfile

class UserSchema(Schema):
    name   = fields.String()
    age    = fields.Integer()
    email  = fields.Email()
    money  = fields.Number()
    class Meta:
        fields = ["name","age","money","email","info"]
        ordered = True # 转换成有序字典

def index():
    """序列化"""
    """多个模型数据的序列化"""
    user1 = User(name="xiaoming", password="123456", age=15, email="333@qq.com", money=31.50)
    user2 = User(name="xiaohong", password="123456", age=16, email="333@qq.com", money=31.50)
    user3 = User(name="xiaopang", password="123456", age=17, email="333@qq.com", money=31.50)
    data_list = [user1,user2,user3]
    data1 = UserSchema(many=True).dumps(data_list)
    print(type(data1),data1)
    return "ok"

构造器嵌套使用

application.apps.marsh.views,代码:

from marshmallow import Schema,fields
from application.apps.users.models import User,UserProfile
class UserProfileSchema(Schema):
    education = fields.Integer()
    middle_school = fields.String()

class UserSchema(Schema):
    name   = fields.String()
    age    = fields.Integer()
    email  = fields.Email()
    money  = fields.Number()
    info   = fields.Nested(UserProfileSchema,only=["middle_school"])
    class Meta:
        fields = ["name","age","money","email","info"]
        ordered = True # 转换成有序字典

def index():
    """序列化"""
    """序列化嵌套使用"""
    user1 = User(name="xiaoming", password="123456", age=15, email="333@qq.com", money=31.50)
    user1.info = UserProfile(
        education=3,
        middle_school="北京师范学院附属中学白沙路分校"
    )
    data = UserSchema().dump(user1)
    data = UserSchema().dumps(user1)
    print(data)
    return "ok"

基于Schema完成数据反序列化转换

代码：

from marshmallow import Schema, fields, validate, ValidationError,post_load
class UserSchema2(Schema):
    name = fields.String()
    sex = fields.String()
    age = fields.Integer(missing=18)
    email = fields.Email()
    mobile = fields.String()

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

def index():
    user_data = {"mobile":"1331345635","name": "xiaoming", "email": "xiaoming@qq.com","sex":"abc"}
    us2 = UserSchema2()
    result = us2.load(user_data)
    print(result)  # ==> <User xiaoming>
    return "ok"

反序列化时转换/忽略部分数据

from marshmallow import Schema, fields, validate, ValidationError,post_load
class UserSchema2(Schema):
    name = fields.String()
    sex = fields.String()
    age = fields.Integer(missing=18)
    email = fields.Email()
    mobile = fields.String(required=True)

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

def index():
    user_data = {"name": "xiaoming","sex":"abc"}
    us2 = UserSchema2()
    result = us2.load(user_data,partial=True)
    print(result)  # ==> <User xiaoming>
    return "ok"

设置字段只在序列化或反序列化阶段才启用

from marshmallow import Schema, fields, validate, ValidationError,post_load
class UserSchema2(Schema):
    name = fields.String()
    sex = fields.Integer(validate=validate.OneOf([0,1,2]))
    age = fields.Integer(missing=18)
    email = fields.Email()
    mobile = fields.String()
    password = fields.String(load_only=True) # 设置当前字段为只写字段 "write-only"，在反序列化启用
	created_time = fields.DateTime(dump_only=True) # 相当于只读字段 "read-only"
    
    @post_load
    def post_load(self, data, **kwargs):
        return User(**data)

def index():
    user_data = {"name": "xiaoming","password":"123456","sex":1}
    us2 = UserSchema2()
    # 反序列化
    result = us2.load(user_data)
    print(result)  # ==> <User xiaoming>
    # 序列化
    us3 = UserSchema2(only=["sex","name","age"]) # 限制处理的字段
    result2 = us3.dump(result)
    print(result2)
    return "ok"

反序列化阶段的钩子方法

post_dump([fn，pass_many，pass_original]) 注册要在序列化对象后调用的方法，它会在对象序列化后被调用。
post_load([fn，pass_many，pass_original]) 注册反序列化对象后要调用的方法，它会在验证数据之后被调用。
pre_dump([fn，pass_many]) 注册要在序列化对象之前调用的方法，它会在序列化对象之前被调用。
pre_load([fn，pass_many]) 在反序列化对象之前，注册要调用的方法，它会在验证数据之前调用。

from marshmallow import Schema, fields, validate, ValidationError,post_load,post_dump
class UserSchema2(Schema):
    name = fields.String()
    sex = fields.Integer(validate=validate.OneOf([0,1,2]))
    age = fields.Integer(missing=18)
    email = fields.Email()
    mobile = fields.String()
    password = fields.String(load_only=True) # 设置当前字段为只写字段，只会在反序列化阶段启用

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

    @post_dump
    def post_dump(self,data, **kwargs):
        data["mobile"] = data["mobile"][:3] +"*****"+ data["mobile"][-3:]
        return data

def index():
    user_data = {"name": "xiaoming","password":"123456","sex":1,"mobile":"133123454656"}
    us2 = UserSchema2()
    # 反序列化
    result = us2.load(user_data)
    print(result)  # ==> <User xiaoming>
    # 序列化
    us3 = UserSchema2(only=["sex","name","age","mobile"]) # 限制处理的字段
    result2 = us3.dump(result)
    print(result2)
    return "ok"

反序列化阶段对数据进行验证

基于内置验证器进行数据验证

内置验证器	描述
validate.`Email`(*, error)	邮箱验证
validate.`Equal`(comparable, *, error)	判断值是否相等
validate.`Length`(min, max, *, equal, error)	值长度/大小验证
validate.`OneOf`(choices, labels, *, error)	选项验证
validate.`Range`([min, max])	范围验证
validate.`Regexp`(regex, bytes, Pattern][, flags])	正则验证
validate.`URL`(*, relative, schemes, Set[str]]] = None, …)	验证是否为URL

代码：

from marshmallow import Schema, fields, validate, ValidationError,post_load
class UserSchema3(Schema):
    name = fields.String(required=True)
    sex = fields.String(required=True,error_messages={"required":"对不起，permission必须填写"})
    age = fields.Integer(missing=18,validate=validate.Range(min=18,max=40,error="年龄必须在18-40之间！")) # 限制数值范围
    email = fields.Email(error_messages={"invalid":"对不起，必须填写邮箱格式！"})
    mobile = fields.String(required=True, validate=validate.Regexp("^1[3-9]\d{9}$",error="手机号码格式不正确"),error_messages={"Regexp":"手机格式不正确"})

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

def index3():
    user_data = {"mobile":"1331345635","name": "xiaoming","age":40, "email": "xiaoming@qq.com","sex":"abc"}
    us2 = UserSchema3()
    result = us2.load(user_data)
    result2 = us2.dumps(result)
    print(result)
    print(result2)
    return "ok"

自定义验证方法

from marshmallow import Schema, fields, validate,validates, ValidationError,post_load,validates_schema
class UserSchema4(Schema):
    name = fields.String(required=True)
    sex = fields.String(required=True,error_messages={"required":"对不起，permission必须填写"})
    age = fields.Integer(missing=18,validate=validate.Range(min=18,max=40,error="年龄必须在18-40之间！")) # 限制数值范围
    email = fields.Email(error_messages={"invalid":"对不起，必须填写邮箱格式！"})
    mobile = fields.String(required=True, validate=validate.Regexp("^1[3-9]\d{9}$",error="手机号码格式不正确"),error_messages={"Regexp":"手机格式不正确"})
    password = fields.String(required=True, load_only=True)
    password2 = fields.String(required=True, allow_none=True)
    
    @post_load
    def make_user_obj(self, data, **kwargs):
        return User(**data)

    @validates("name")
    def validate_name(self,data,**kwargs):
        print("name=%s" % data)
        if data == "root":
            raise ValidationError({"对不起,root用户是超级用户!您没有权限注册！"})

        # 必须有返回值
        return data

    @validates_schema
    def validate(self,data,**kwargs):
        print(data)
        if data["password"] != data["password2"]:
            raise ValidationError("密码和确认密码必须一样！")

        data.pop("password2")
        return data

def index():
    user_data = {"password":"12345","password2":"123456","mobile":"13313345635","name": "root1","age":40, "email": "xiaoming@qq.com","sex":"abc"}
    us2 = UserSchema4()
    result = us2.load(user_data)
    print(result)
    return "ok"

模型构造器(ModelSchema)

官方文档：https://github.com/marshmallow-code/marshmallow-sqlalchemy

https://marshmallow-sqlalchemy.readthedocs.io/en/latest/

注意：flask_marshmallow在0.12.0版本以后已经移除了ModelSchema和TableSchema这两个模型构造器类，官方转而推荐了使用SQLAlchemyAutoSchema和SQLAlchemySchema这2个类，前后两者用法类似。

创建

from marshmallow_sqlalchemy import SQLAlchemySchema,SQLAlchemyAutoSchema,auto_field
class UserSchema5(SQLAlchemySchema):
    # auto_field的作用,设置当前数据字段的类型和选项声明自动从模型中对应的字段中提取
    # name = auto_field()
    # 此处，数据库中根本没有username，需要在第一个参数位置，声明当前数据字典的类型和选项声明从模型的哪个字段提取的
    username = auto_field("name",dump_only=True)
    # 可以在原字段基础上面，增加或者覆盖模型中原来的声明
    created_time = auto_field(format="%Y-%m-%d")
    # 甚至可以声明一些不是模型的字段
    token = fields.String()
    class Meta:
        model = User
        fields = ["username","created_time","token"]

def index5():
    """单个模型数据的序列化处理"""
    from datetime import datetime
    user1 = User(
        name="xiaoming",
        password="123456",
        age=16,
        email="333@qq.com",
        money=31.50,
        created_time= datetime.now(),
    )
    user1.token = "abc"
    # 把模型对象转换成字典格式
    data1 = UserSchema5().dump(user1)
    print(type(data1),data1)
    return "ok"

"""SQLAlchemySchema使用起来，虽然比上面的Schema简单许多，但是还是需要给小转换的字段全部统一写上才转换这些字段
，如果不想编写字段信息，直接从模型中复制，也可以使用SQLAlchemyAutoSchema。"""
class UserSchema6(SQLAlchemyAutoSchema):
    token = fields.String()
    class Meta:
        model = User
        include_fk = False # 启用外键关系
        include_relationships = False # 模型关系外部属性
        fields = ["name","created_time","info","token"] # 如果要全换全部字段，就不要声明fields或exclude字段即可
        sql_session = db.session

def index():
    """单个模型数据的序列化处理"""
    from datetime import datetime
    user1 = User(
        name="xiaoming",
        password="123456",
        age=16,
        email="333@qq.com",
        money=31.50,
        created_time= datetime.now(),
        info=UserProfile(position="助教")
    )
    # 把模型对象转换成字典格式
    user1.token="abcccccc"
    data1 = UserSchema6().dump(user1)
    print(type(data1),data1)
    return "ok"