使用Flask和Pydantic实现参数验证

1 简介

Pydantic是一个用于数据验证和解析的 Python 库，版本2的性能有较大提升，很多框架使用Pydantic做数据校验。

# 官方参考文档
https://docs.pydantic.dev/latest/# Github地址
https://github.com/pydantic/pydantic

2 使用Flask自定义参数校验

2.1 结构

2.2 源代码

app_result.py

import jsonfrom flask import jsonifydef result(data: object = None, code: int = 200, msg: str = "success"):"""Set the result of web:param data: It is data of status:param code: It is code of status:param msg: It is a message of status:return: dict"""# 将对象转化为json字符串data_json_str = json.dumps(data, ensure_ascii=False, default=lambda obj: obj.__dict__)# 将json字符串转化为json字典data_json_dict = json.loads(data_json_str)# 设置返回值字典result_dict = {"code": code,"msg": msg,"data": data_json_dict}return jsonify(result_dict)

validate_request.py

from flask import request
from pydantic import ValidationErrorfrom entity.app.app_result import resultdef validate_form(form_object):# 设置含有请求参数的装饰器def inner(fun):def inner_sub():try:# 获取Content-Typecontent_type = request.content_typeif content_type is None:return result(code=400, msg="No parameters")# 验证参数form_instance = None# 获取 “application/json” 请求if "application/json" in content_type:# 获取json字典数据parameters = request.get_json(silent=True)# 将字典转化为对象form_instance = form_object(**parameters)else:# 获取 "application/x-www-form-urlencoded" 或者 "multipart/form-data" 请求等parameters = request.form.to_dict()# 将字典转化为对象form_instance = form_object(**parameters)# print(parameters)return fun(form_instance)except ValidationError as e:print(e.json())# 存储错误参数列表parameters_invalid = list()# 处理参数for error in e.errors():parameters_invalid.append({"attr": error.get("loc")[0],"msg": error.get("msg")})return result(parameters_invalid, code=400, msg="The parameter is invalid")# 更改视图名称inner_sub.__name__ = fun.__name__return inner_subreturn inner

data_form.py

from typing import Optionalfrom pydantic import BaseModel, Field, field_validatorclass DataForm(BaseModel):# Python中"..."3个点是Ellipsis对象，主要用于切片操作和类型注解‌# pydantic中Field的"..."3个点表示必填项name: str = Field(..., title="The name of the item", description="deldldlddl", max_length=5)desc: Optional[str] = Field(..., title="The description of the item", max_length=2)price: float = Field(..., title="The price of the item", ge=0.01)"""# 如果不使用Field设置验证参数，可以使用@field_validator注解# 自定义验证器@field_validator('price')# cls是一个用于类方法的参数，代表类本身def age_must_be_positive(cls, v):if v < 0:raise ValueError('年龄不能为负数')return v"""

index.py

from flask import Flask, Blueprintfrom test.data_form import DataForm
from entity.form.validate_request import validate_formapp = Flask(__name__)# 构建蓝本
admin = Blueprint("admin", __name__)@admin.route("/info")
@validate_form(DataForm)
def info(info_form: DataForm):print(info_form.name)print(info_form.desc)return "Hello Admin"@admin.route("/msg")
@validate_form(DataForm)
def msg(msg_form: DataForm):print(msg_form)return "Hello Admin"# 注册蓝本
app.register_blueprint(admin, url_prefix='/admin')if __name__ == '__main__':app.run(host='0.0.0.0', debug=True)

2.3 返回值