Marshmallow简介
标签: Marshmallow简介 博客 51CTO博客
2023-07-27 18:24:12 173浏览
模块简介
Marshmallow,中文译作:棉花糖.是一个轻量级的数据格式转换的模板,常用于将复杂的orm模型对象与python原生数据类型之间相互转换.
官方文档:https://marshmallow.readthedocs.io/en/latest/
使用背景
在使用RESTful API进行开发的过程中,序列化与反序列化是绕不开的一环,而marshmallow在序列化与反序列化,数据验证和转换,数据格式化,嵌套字段和数据关联,数据过滤和选择具有很好的功能表现
具体使用
首先在pycharm上新建一个flask项目,结构如下:
marshmallow的使用,将从下面几个方面展开,在开始之前,首先需要一个用于序列化和反序列化的类,在model文件里定义
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常用属性数据类型
schema数据类型的常用通用属性
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
,并具有默认值为当前日期和时间,以便在更新用户信息时使用。
序列化测试
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)
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()
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
先生成数据库结构
然后以此执行flask db init , flask db migrate, flask db upgrade
插入初始数据
直接执行data_init.py,运行完成后查看数据插入成功
运行项目后请求url:10.0.40.99:5000/api/users
返回:
反序列化测试
尝试使用错误的信息提交
反序列时转换/忽略部分数据
按照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一共提供了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}"})
执行返回
反序列化阶段对数据进行验证
基于内置验证器进行数据验证
我在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框架,它们其中有很多模板,功能,风格都是相互借鉴的,只要明白其中一个框架,另外一个框架其实非常容易理解,我们不仅需要有深度了解一个框架的毅力,更要有横向扩展学习的眼界.
好博客就要一起分享哦!分享海报
此处可发布评论
评论(0)展开评论
展开评论