flask中的marshmallow使用

您所在的位置：网站首页 › marshmallow头像图片 › flask中的marshmallow使用

flask中的marshmallow使用

2024-07-11 23:02| 来源: 网络整理| 查看: 265

一、介绍

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

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

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 app 三、使用

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): # 基本构造器(Schema) '''转换用户信息与模型无关 fields设置字段类型 ''' 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="[email protected]", money=31.50 ) # print(user1) # 把模型对象转换成字典格式 data1 = UserSchema().dump(user1) print(type(data1),data1) # 把模型对象转换成json字符串格式 data2 = UserSchema().dumps(user1) # dumps把任意对象序列化成一个str 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”: “用户名为必填项。”}

序列化多个： 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="[email protected]", money=31.50) user2 = User(name="xiaohong", password="123456", age=16, email="[email protected]", money=31.50) user3 = User(name="xiaopang", password="123456", age=17, email="[email protected]", money=31.50) data_list = [user1,user2,user3] data1 = UserSchema(many=True).dumps(data_list) # dumps把任意对象序列化成一个str many=True 允许多个模型序列化 print(type(data1),data1) # [{"name": "xiaoming", "age": 4, "money": 31.5, "email": "[email protected]", "info": null}, {"name": "xiaohui", "age": 5, "money": 66.5, "email": "[email protected]", "info": null}, {"name": "xiaobai", "age": 6, "money": 88.5, "email": "[email protected]", "info": null}] return "ok" 构造器嵌套使用： 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"]) # nested外键类型 only只显示某个字段 class Meta: fields = ["name","age","money","email","info"] ordered = True # 转换成有序字典 def index(): """序列化""" """序列化嵌套使用""" user1 = User(name="xiaoming", password="123456", age=15, email="[email protected]", money=31.50) user1.info = UserProfile( education=3, middle_school="qwq学校" ) data = UserSchema().dump(user1) # order排序之后的字典 print(data) # OrderedDict([('name', 'xiaoming'), ('age', 15), ('money', 31.5), ('email', '[email protected]'), ('info', {'middle_school': 'qwq学校'})]) data1 = UserSchema().dumps(user1) # 对象转字符串 print(data1) # {"name": "xiaoming", "age": 15, "money": 31.5, "email": "[email protected]", "info": {"middle_school": "qwq\u5b66\u6821"}} return "ok" 基于Schema完成数据反序列化转换： class UserSchema2(Schema): name = fields.String(required=True,error_messages={"required":"必须填写"}) # required=True 必须要传的参数，没有则error sex = fields.String() age = fields.Integer(missing=18) # 反序列化时设置默认值 email = fields.Email(error_messages={'invalid':"对不起，必须正确填写邮箱格式化！"}) mobile = fields.String() @post_load # 反序列化成一个对象，给schema的一个方法加上post_load装饰器 def post_load(self, data, **kwargs): return User(**data) def index(): user_data = {"mobile":"1331345635", "email": "[email protected]","sex":"abc"} us2 = UserSchema2() result = us2.load(user_data,partial=True) # 默认序列化器必须传递所有required的字段，否则会抛出验证异常。但是我们可以使用partial=True，来避免报错 print(result) # ==> 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) # 默认序列化器必须传递所有required的字段，否则会抛出验证异常。但是我们可以使用partial参数来允许部分字段更新 print(result) # ==> return "ok" 设置字段只在序列化或反序列化阶段才启用： 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" 只会在反序列化阶段启用 # password = fields.String(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) # ==> # 序列化 us3 = UserSchema2(only=["sex","name","age","password"]) # 限制处理的字段 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) # ==> # 序列化 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": "[email protected]","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": "[email protected]","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个类，前后两者用法类似。

class UserSchema(SQLAlchemyAutoSchema): class Meta: model = 模型类名 # table = models.Album.__table__ include_relationships = True # 输出模型对象时同时对外键，是否也一并进行处理 include_fk = True # 序序列阶段是否也一并返回主键 load_instance = True # 反序列化阶段时，直接返回模型对象 sql_session = db.session # 数据库连接会话对象 # fields= ["id","name"] # 启动的字段列表 exclude = ["id","name"] # 排除字段列表

例子：

from marshmallow_sqlalchemy import SQLAlchemySchema,SQLAlchemyAutoSchema,auto_field from marshmallow import Schema, fields from application.apps.user.models import User, UserProfile, db from marshmallow import Schema, fields, validate, ValidationError,post_load class UserSchema5(SQLAlchemySchema): # auto_field的作用,设置当前数据字段的类型和选项声明自动从模型中对应的字段中提取 # name = auto_field() # 此处，数据库中根本没有username，需要在第一个参数位置，声明当前数据字典的类型和选项声明从模型的哪个字段提取的 username = auto_field("name",dump_only=True) # 可以在原字段基础上面，增加或者覆盖模型中原来的声明 created_time = auto_field(format="%Y-%m-%d") # format 设置时间格式 # 甚至可以声明一些不是模型的字段 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="[email protected]", money=31.50, created_time= datetime.now(), ) user1.token = "abc" # 把模型对象转换成字典格式 data1 = UserSchema5().dump(user1) print(type(data1),data1) # {'username': 'xiaoming', 'token': 'abc', 'created_time': '2020-12-02'} 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="[email protected]", money=31.50, created_time= datetime.now(), info=UserProfile(position="助教") ) # 把模型对象转换成字典格式 user1.token="abcccccc" data1 = UserSchema6().dump(user1) print(type(data1),data1) # class 'dict'> {'info': , 'created_time': '2020-12-02T17:47:21.821940', 'token': 'abcccccc', 'name': 'xiaoming'} return "ok"

【本文地址】

flask中的marshmallow使用

flask中的marshmallow使用

今日新闻

推荐新闻