使用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 json

from flask import jsonify


def 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 ValidationError

from entity.app.app_result import result


def validate_form(form_object):
    # 设置含有请求参数的装饰器
    def inner(fun):
        def inner_sub():
            try:
                # 获取Content-Type
                content_type = request.content_type
                if 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_sub
    return inner

data_form.py

from typing import Optional

from pydantic import BaseModel, Field, field_validator


class 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, Blueprint

from test.data_form import DataForm
from entity.form.validate_request import validate_form

app = 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 返回值