flask-SQLAlchemy-pydantic实现数据校验的增删改查操作_sqlalchemy pydantic

from typing import Optional
 
from flask import Flask, jsonify, views, request
from flask_sqlalchemy import SQLAlchemy
from flask_pydantic import validate
 
from pydantic import BaseModel
 
from demo import Role, to_json
 
app = Flask(__name__)
 
# 通过 app 配置数据库信息
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:149789@127.0.0.1:3306/sqlalchemy_test'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
 
# 创建数据库实例
db = SQLAlchemy(app)
 
 
class RolePageModel(BaseModel):
    page: Optional[int] = None
    size: Optional[int] = None
 
 
class RoleCreateModel(BaseModel):
    data: str
    age: int
 
 
class RoleUpdateModel(BaseModel):
    id: int
    data: Optional[str] = None
    age: int
 
 
class RoleDeleteModel(BaseModel):
    id: int
 
 
class RoleView(views.MethodView):
    @validate()
    def get(self, query: RolePageModel):
        # 获取角色列表
        page = query.page
        size = query.size
        if not page:
            page = 1
        if not size:
            size = 2
        if size > 2:
            size = 2
        # 获取角色列表
        try:
            all_obj = db.session.query(Role).paginate(page, size).items
        except Exception:
            return jsonify({"status": 404, "msg": "没有内容了"})
        ret_json = to_json(all_obj)
        return jsonify(ret_json)
 
    @validate()
    def post(self, body: RoleCreateModel):
        # 创建一个角色
        data = body.data
        age = body.age
        r1 = Role(data=data, age=age)
        db.session.add(r1)
        db.session.commit()
        return jsonify({"code": 201, "msg": "添加成功", "role": r1.to_dict()})
 
    @validate()
    def patch(self, query: RoleUpdateModel):
        # 更新一个角色
        id = query.id
        age = query.age
        r1 = db.session.query(Role).filter(Role.id == id).first()
        if not r1:
            return jsonify({"status": 404, "msg": "内容没找到"})
        if query.data:
            r1.data = query.data
        r1.age = age
        db.session.commit()
        return jsonify({"code": 201, "msg": "更新成功", "role": r1.to_dict()})
 
    @validate()
    def delete(self, body: RoleDeleteModel):
        # 删除一个角色
        # r1 = db.session.query(Role).filter(Role.id == id).first()
        id = body.id
        # get方法，找不到返回None
        r1 = db.session.query(Role).get(id)
        if not r1:
            return jsonify({"status": 404, "msg": "内容没找到"})
        db.session.delete(r1)
        db.session.commit()
        return jsonify({"code": 204, "msg": "删除成功"})
 
 
app.add_url_rule("/role", view_func=RoleView.as_view("role"))
 
if __name__ == '__main__':
    app.run()

flask-pydantic参考链接：https://github.com/bauerji/flask_pydantic