Marshmallow简介-摩杜云开发者社区

模块简介

Marshmallow,中文译作:棉花糖.是一个轻量级的数据格式转换的模板,常用于将复杂的orm模型对象与python原生数据类型之间相互转换.

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

使用背景

在使用RESTful API进行开发的过程中,序列化与反序列化是绕不开的一环,而marshmallow在序列化与反序列化,数据验证和转换,数据格式化,嵌套字段和数据关联,数据过滤和选择具有很好的功能表现

具体使用

首先在pycharm上新建一个flask项目,结构如下:

Marshmallow简介_序列化

marshmallow的使用,将从下面几个方面展开,在开始之前,首先需要一个用于序列化和反序列化的类,在model文件里定义

extension.py

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

model.py内容

from datetime import datetime

from extension import db


class Base(db.Model):
    """基类"""
    # 作为父类被继承，不会被创建成表（抽象）
    __abstract__ = True
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)

    def save(self):
        db.session.add(self)
        db.session.commit()

    def delete(self):
        db.session.delete(self)
        db.session.commit()

    def update(self):
        print(self)
        db.session.commit()


class User(Base):
    username = db.Column(db.String(255), index=True, comment="用户名")
    password = db.Column(db.String(255), comment="登录密码")
    mobile = db.Column(db.String(15), index=True, comment="手机号码")
    sex = db.Column(db.Boolean, default=True, comment="性别")
    email = db.Column(db.String(255), index=True, comment="邮箱")
    created_time = db.Column(db.DateTime, default=datetime.now, comment="创建时间")
    updated_time = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now, comment="更新时间")

    def __init__(self, username, password, mobile, sex, email):
        self.username = username
        self.password = password
        self.mobile = mobile
        self.sex = sex
        self.email = email

Schema

要对model中的User类进行序列化和反序列化,首先要创建一个与之对应Schema类,负责实现User类的序列化,反序列化和数据校验等

schema常用属性数据类型

Marshmallow简介_restful_02

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

Marshmallow简介_序列化_03

modelschema.py 填入以下内容

import datetime

from marshmallow import Schema, fields, validate


class BaseSchema(Schema):
    """
    基类结构
    """
    id = fields.Int(dump_only=True, description="唯一标识符")


class UserSchema(BaseSchema):
    """
    User结构
    """
    username = fields.Str(description="用户名", required=True)
    password = fields.Str(description="登录密码", required=True, load_only=True)
    mobile = fields.Str(description="手机号码", required=True, validate=validate.Regexp(r'^\d{11}$',
                                                                                    error='Invalid phone number.'))
    sex = fields.Boolean(description="性别", required=True, validate=validate.OneOf([0, 1]))
    email = fields.Email(description="邮箱", required=True)
    created_time = fields.DateTime(description="创建时间", allow_none=True,
                                   default=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
    updated_time = fields.DateTime(description="更新时间", allow_none=True,
                                   default=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))

在modelschema中我们定义了一个

UserSchema类，它继承了 BaseSchema。 UserSchema用于序列化和反序列化用户的信息。

以下是各字段及其属性的解释：

id：一个整数字段，用于唯一标识用户，dump_only=True表示在序列化（转换为JSON等格式）输出时显示该字段，但在反序列化（加载）数据时不需要考虑它。
username：一个字符串字段，表示用户的用户名。它被标记为required=True，意味着在反序列化数据（即从JSON等格式转换为对象）时必须包含该字段。
password：一个字符串字段，表示用户的登录密码。它被标记为required=True，但具有load_only=True属性，这意味着在序列化输出（即转换为JSON等格式）时不会显示该字段，因为密码是敏感信息，不应该暴露给客户端。
mobile：一个字符串字段，表示用户的手机号码。它被标记为required=True，并使用正则表达式进行验证，以确保它符合11位数字的模式。
sex：一个布尔字段，表示用户的性别。它被标记为required=True，并使用validate.OneOf([0, 1])进行验证，意味着它必须是0或1，分别表示性别的两种可能值。
email：一个字符串字段，表示用户的电子邮箱地址。它被标记为required=True，并进行验证以确保它是有效的电子邮箱地址。
created_time：一个日期时间字段，表示用户的创建时间。它被标记为allow_none=True，意味着在反序列化数据时可以将其设置为None，如果未提供创建时间，则默认为当前日期和时间。
updated_time：一个日期时间字段，表示用户的最后更新时间。类似于created_time，它被标记为allow_none=True，并具有默认值为当前日期和时间，以便在更新用户信息时使用。

序列化测试

app.py

import os

from flask import Flask
from flask_restful import Api
from extension import db
from Http.JsonTest import UsersList, Users

app = Flask(__name__)
# 尤其在涉及（flask-WTF）登陆页面提交敏感信息时，一定要设置密钥
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY') or 'westos secret key'
db.init_app(app)

app.config['SQLALCHEMY_DATABASE_URI'] = "sqlite:///test.sqlite"
# 是否自动提交、
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False  # 该配置默认开启，会引入额外性能开销且未来将不被支持，故禁用
app.config['SQLALCHEMY_COMMIT_ON_TEARDOM'] = True
api = Api(app)

# 注册api路由
api.add_resource(UsersList, '/api/users')
api.add_resource(Users, '/api/users/<int:user_id>')
if __name__ == '__main__':
    app.run(debug=True, host="10.0.40.99", port=5000)

manager.py

from flask_migrate import Migrate
from app import app
from extension import db
migrate = Migrate(app, db, render_as_batch=True)

# set FLASK_APP=manager
# export FLASK_APP="manager"
# flask db init
# flask db migrate
# flask db upgrade

data_init.py

from model import *


def insert_user():
    from app import app
    with app.app_context():
        user1 = User(
            username="xiaoming1号",
            password='12345',
            mobile="13312345677",
            sex=True,
            email="133123456@qq.com"
        )

        user2 = User(
            username="xiaoming2号",
            password='12345',
            mobile="13312345677",
            sex=True,
            email="133123456@qq.com"
        )
        user1.save()
        user2.save()


insert_user()

JsonTest.py

from flask_restful import Resource
from modelschema import *
from model import *
from flask import jsonify, request


class UsersList(Resource):
    """
    获取所有用户信息
    """

    def get(self):
        print(self)
        u_s = UserSchema()
        users = User.query.all()
        return jsonify({"success": True, "msg": u_s.dump(users, many=True)})


class Users(Resource):
    """
    单个用户操作
    """

    def post(self, user_id):
        print(self)
        u_s = UserSchema()
        if User.query.filter(User.id == user_id).first():
            return jsonify({"success": False, "msg": "数据已存在,无法处理"})
        try:
            args = request.json
            params = {}
            for i in args:
                params[i] = args[i]
            result = u_s.load(params)
        except Exception as err:
            return jsonify({"success": False, "msg": f"{err}"})

        try:
            objuser = User(**result)
            objuser.save()
            # 入库成功后返回user id
            user_id = u_s.dump(objuser)["id"]
            return jsonify({"success": True, "msg": user_id})
        except Exception as E:
            return jsonify({"success": False, "msg": "入库失败:{}".format(str(E))})

在上面的代码中,UserList是一个继承自Resource的资源类,它用于处理获取所有用户信息的请求。

在get方法中,首先创建了UserSchema的实例u_s,UserSchema是用于格式化用户数据的Schema类.然后通过User.query.all()查询数据库,获取所有用户信息,并将查询结果存储在users中。

接下来,调用u_s.dump(users, many=True)对查询结果进行序列化和格式化,将数据库查询结果转换为JSON格式的数据.u_s.dump()方法的第一个参数是要序列化的数据对象,第二个参数many=True表示要处理多个数据对象,因为users是一个包含多个用户信息的列表.

最后,使用jsonify()函数将格式化后的数据以JSON格式返回给客户端.

requirements.txt 依赖包文件

alembic==1.11.1
aniso8601==9.0.1
blinker==1.6.2
click==8.1.6
colorama==0.4.6
Flask==2.3.2
Flask-Migrate==3.1.0
Flask-RESTful==0.3.10
Flask-SQLAlchemy==2.5.1
greenlet==2.0.2
importlib-metadata==6.8.0
importlib-resources==6.0.0
itsdangerous==2.1.2
Jinja2==3.1.2
Mako==1.2.4
MarkupSafe==2.1.3
marshmallow==3.20.1
packaging==23.1
pytz==2023.3
six==1.16.0
SQLAlchemy==1.4.36
typing_extensions==4.7.1
Werkzeug==2.3.6
zipp==3.16.2

先生成数据库结构

Marshmallow简介_反序列化_04

然后以此执行flask db init ， flask db migrate， flask db upgrade

插入初始数据

直接执行data_init.py,运行完成后查看数据插入成功

Marshmallow简介_反序列化_05

运行项目后请求url:10.0.40.99:5000/api/users

Marshmallow简介_restful_06

反序列化测试

Marshmallow简介_marshmallow_07

尝试使用错误的信息提交

Marshmallow简介_restful_08

Marshmallow简介_反序列化_09

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

按照restful架构风格的要求,更新数据使用http方法中的put或patch方法,使用put方法时需要把完整的数据全部传给服务器,使用patch方法时,只要把需要改动的部分数据传给服务器即可.因此,当使用patch方法时,传入数据存在无法通过marshmallow数据校验的风险,为了避免这种情况,需要借助partial loading功能,实现parital loading只要在schema中增加一个partial参数即可

JsonTest.py加入下面内容

class Users(Resource):
    """
    单个用户操作
    """

    def post(self, user_id):
        print(self)
        u_s = UserSchema()
        if User.query.filter(User.id == user_id).first():
            return jsonify({"success": False, "msg": "数据已存在,无法处理"})
        try:
            args = request.json
            params = {}
            for i in args:
                params[i] = args[i]
            result = u_s.load(params)
        except Exception as err:
            return jsonify({"success": False, "msg": f"{err}"})

        try:
            objuser = User(**result)
            objuser.save()
            # 入库成功后返回user id
            user_id = u_s.dump(objuser)["id"]
            return jsonify({"success": True, "msg": user_id})
        except Exception as E:
            return jsonify({"success": False, "msg": "入库失败:{}".format(str(E))})

    def patch(self, user_id):
        print(self)
        u_s = UserSchema()
        userobj = User.query.filter(User.id == user_id).first()
        if not userobj:
            return jsonify({"success": False, "msg": "数据不存在,无法处理"})
        try:
            args = request.json
            params = {}
            for i in args:
                params[i] = args[i]
            result = u_s.load(params, partial=True)   # 关键地方
        except Exception as err:
            return jsonify({"success": False, "msg": f"{str(err)}"})

        try:
            for key, value in params.items():
                if hasattr(userobj, key):
                    setattr(userobj, key, value)
            userobj.update()
            # 更新成功后返回用户信息
            user_id = u_s.dump(userobj)
            return jsonify({"success": True, "msg": user_id})
        except Exception as E:
            return jsonify({"success": False, "msg": "更新失败:{}".format(str(E))})

加了一个patch函数

测试返回:

Marshmallow简介_marshmallow_10

反序列化阶段的钩子方法

marshmallow一共提供了4个钩子方法:

pre_dump([fn, pass_many])注册要再序列化对象之前调用的方法,它会在序列化对象之前被调用.

pre_load([fn, pass_many])在序列化对象之前,注册要调用的方法,它会在验证数据之前调用.

post_dump([fn, pass_many, pass_original])注册要在序列化对象后调用的方法,它会在对象序列化后被调用.

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

这个要看自己项目需要,不是什么东西看到了就往自己项目里添加,这样会造成在后续开发和维护中的一些困难

我用post_load举个例子

modelschema.py 添加以下内容

from marshmallow import Schema, fields, validate, post_load
from werkzeug.security import generate_password_hash
 @post_load
    def post_load(self, data, **kwargs):
        """
        反序列化的后置钩子,会在数据验证之后执行
        :param data:
        :param kwargs:
        :return:
        """
        print(data, kwargs)
        data["password"] = generate_password_hash(data["password"])
        return User(**data)

JsonTest.py做修改

class Users(Resource):
    """
    单个用户操作
    """

    def post(self, user_id):
        print(self)
        u_s = UserSchema()
        if User.query.filter(User.id == user_id).first():
            return jsonify({"success": False, "msg": "数据已存在,无法处理"})
        try:
            args = request.json
            params = {}
            for i in args:
                params[i] = args[i]
            userobj = u_s.load(params)
            userobj.save()
            # 入库成功后返回user id
            user_id = u_s.dump(userobj)["id"]
            return jsonify({"success": True, "msg": user_id})
        except Exception as err:
            return jsonify({"success": False, "msg": f"{err}"})

执行返回

Marshmallow简介_反序列化_11

Marshmallow简介_restful_12

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

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

Marshmallow简介_restful_13

我在modelschema.py文件中有使用部分验证器,这个可以自己多做测试看看

自定义验证方法

modelschema.py添加修改下面内容

class UserSchema(BaseSchema):
    """
    User结构
    """
    username = fields.Str(description="用户名", required=True)
    password = fields.Str(description="登录密码", required=True, load_only=True)
    mobile = fields.Str(description="手机号码", required=True, validate=validate.Regexp(r'^\d{11}$',
                                                                                    error='Invalid phone number.'))
    sex = fields.Boolean(description="性别", required=True, validate=validate.OneOf([0, 1]))
    email = fields.Email(description="邮箱", required=True)
    created_time = fields.DateTime(description="创建时间", allow_none=True,
                                   default=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
    updated_time = fields.DateTime(description="更新时间", allow_none=True,
                                   default=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))

    # 针对单个指定字段的值进行验证
    @validates("mobile")
    def validate_mobile(self, value):
        if value == "13312345678":
            raise ValueError("手机号码已被注册!")

    # 针对多个字段的验证
    @validates_schema
    def validate(self, data, **kwargs):
        print(data, kwargs)
        if len(data["username"]) or len(data["password"]) > 255:
            raise ValueError("username or password must be less than 255 characters")

这个自定义验证我就不测试了,你们可以自己试试看

小结

当我们根据marshmallow官方文档上的示例都看完后,有没有发现这个marshmallow跟django drf框架中的serializers很相似呢,是不是很有意思,再横向对比下,比如flask中的url_for和django中的reverse,flask中的jsonify和django中的JsonResponse,flask中的request和django中的request,flask中的蓝图和django中的urlpatterns...其实就python web框架,它们其中有很多模板,功能,风格都是相互借鉴的,只要明白其中一个框架,另外一个框架其实非常容易理解,我们不仅需要有深度了解一个框架的毅力,更要有横向扩展学习的眼界.