Tortoise-ORM集成（1）

查询参数查询参数是URL中?后面的键值对组合格式为key1value1key2value2用于对资源进行「筛选、分页、排序」等辅助操作。eg: /users?name张三age20name姓名、age年龄是查询参数对比维度路径参数查询参数位置URL路径中/items/{item_id}URL后缀?skip0limit10必要性必填默认可选默认可设置必填核心用途标识唯一资源如ID、UID筛选、分页、排序等辅助操作配置工具Path类Query类示例场景/items/1、/users/100/items?limit10、/users?name张三# 1. 字符串查询参数搜索关键词 app.get(/search/items/, summary字符串查询参数校验) def search_items( q: str | None Query( None, # 默认None可选 min_length1, max_length100, patternr^[^#$%^*]$, # 禁止包含特殊字符#$%^* description搜索关键词1-100位禁止包含#$%^* ) ): if not q: return {code: 200, message: 请输入搜索关键词} return {code: 200, message: f搜索「{q}」成功, data: []} # 2. 数值查询参数价格筛选 app.get(/filter/products/, summary数值查询参数校验) def filter_products( min_price: float Query(0.0, ge0.0, description最低价格≥0), max_price: float Query(1000.0, le1000.0, description最高价格≤1000) ): return { code: 200, filter: f价格区间{min_price} - {max_price}, message: 筛选产品成功 } # 3. 别名参数解决前端传参命名规范问题 app.get(/pages/, summaryQuery别名参数示例) def get_pages( page: int Query(1, ge1, description页码), # 后端参数名page_size前端传page-size中划线命名 page_size: int Query(20, ge1, le50, aliaspage-size, description每页条数) ): return { code: 200, page: page, page_size: page_size, message: 分页查询成功 }由此课间Query查询参数是写在def方法括号里可以是关键字分页等多种方式查询也可以进行组合中间用‘’隔开查询参数一般都是选填可有可无如果是必传的情况可以在条件之前加上“...”表示必填如下所示app.get(/users/filter/, summary必选查询参数) def filter_users( # 必选查询参数必须传入username否则报错 username: str Query( ..., # ... 表示必填无默认值 min_length3, max_length20, description用户名必填3-20位字符 ), # 可选查询参数年龄 age: int | None Query(None, ge1, le120, description年龄可选) ): result {code: 200, username: username, message: 筛选用户成功} if age: result[age] age return result请求体前端要一次性传一堆信息比如新增用户用户名、邮箱、密码、年龄这么多参数塞网址里不现实所以把所有数据打包放在请求体body里发给后端这部分打包的数据就叫请求体。BaseModelField提前定义好前端必须传什么、每个数据有什么规则前端传错自动返回提示不用自己写 if 判断校验。# 带字段校验的用户注册模型 class UserCreateWithValidate(BaseModel): 带字段校验的用户注册请求体模型 # 用户名3-20位仅字母/数字/下划线必填 username: str Field( ..., # 必填字段 min_length3, max_length20, patternr^[a-zA-Z0-9_]$, title用户名, description3-20位仅支持字母、数字、下划线, examplezhangsan_123 ) # 邮箱符合邮箱格式必填 email: str Field( ..., patternr^[a-zA-Z0-9._%-][a-zA-Z0-9.-]\.[a-zA-Z]{2,}$, title邮箱, description请输入合法的邮箱地址, examplezstest.com ) # 密码6-20位必填 password: str Field( ..., min_length6, max_length20, title密码, description6-20位字符建议包含字母和数字, example123456a ) # 年龄1-120岁可选默认None age: int | None Field( None, ge1, le120, title年龄, description1-120岁之间, example25 ) # 带校验的注册接口 app.post(/users/register/validate/, summary带字段校验的注册接口) def user_register_validate(user_info: UserCreateWithValidate): return { code: 200, message: 注册成功带字段校验, data: { username: user_info.username, email: user_info.email, age: user_info.age or 未填写 } }Tortoise-ORM 模型定义常用字段类型速查表字段类型Python 类型数据库类型说明fields.IntField(pkTrue)intINT AUTO_INCREMENT自增主键fields.CharField(max_lengthN)strVARCHAR(N)可变长字符串fields.TextField()strTEXT长文本fields.BooleanField(defaultTrue)boolBOOL/TINYINT布尔值fields.DatetimeField(auto_now_addTrue)datetimeDATETIME创建时间自动填充fields.DatetimeField(auto_nowTrue)datetimeDATETIME更新时间自动更新fields.DecimalField(max_digits, decimal_places)DecimalDECIMAL精确小数用于金额fields.FloatField()floatFLOAT浮点数fields.JSONField()dict/listJSONJSON 数据fields.ForeignKeyField()FOREIGN KEY外键一对多fields.OneToOneField()关联模型FOREIGN KEY UNIQUE一对一关系fields.ManyToManyField()关联模型中间表多对多关系实操查task_router.get(/all,summary获取所有任务) async def get_all_tasks( keyword:str|NoneQuery(None,description搜索关键词), sort_by_priority: bool Query(True, description按优先级排序) ): queryTask.all() if keyword: queryquery.filter(Q(title__icontainskeyword)) if sort_by_priority: queryquery.order_by(-priority) else: queryquery.order_by(id) infoawait query return { msg:ok, code:200, data:info }在def 方法里加了query参数做条件查询query Task.all()仅拼接查询条件获取到符合条件的数据之后在进行过滤最后await才执行真正的SQL从数据库拿到任务列表数据。注意不能提前写 Task.all()Task.all()一旦提前 cawait变量变成 list后面filter/order_by会直接报错。增task_router.post(/add,summary添加任务) async def add_task(task:TaskAdd): infoawait Task.create(**dict(task)) return { msg:ok, code:200, data:info }class TaskAdd(BaseModel): user_id:intField( ..., title用户ID, description用户ID, example1 ) title:strField( ..., title任务标题, description任务标题, example学习 ) description:strField( None, title任务描述, description任务描述, example学习Python ) status:intField( 0, title任务状态, description任务状态, example0 ) priority:intField( 1, title任务优先级, description任务优先级, example1 ) due_date:strField( None, title任务截止时间, description任务截止时间, example2023-05-01 ) completed_at:strField( None, title任务完成时间, description任务完成时间, example2023-05-01 ) created_at:strField( None, title任务创建时间, description任务创建时间, example2023-05-01 ) updated_at:strField( None, title任务更新时间, description任务更新时间, example2023-05-01 )在Basemodel中接受前端传过来的参数“...”表示必传None表示可有可无。然后新增核心语句infoawait Task.create(**dict(task))dict(task)将模型对象转为标准字典**是字典解包将键值对平铺传入create方法改task_router.put(/update/{id},summary更新任务) async def update_task(id:int ,task:TaskUpdate): infoawait Task.filter(idid).update(**dict(task)) return { msg:ok, code:200, data:info }直接获取到当前要修改的id的数据进行更新与增加同理在basemodel中获取前端要传的参数与之不同的是basemodel中所有此参数都是None可有可无。对于不存在的id也不会抛出异常也会显示修改成功对于部分修改以状态为例task_router.put(/{id}/status,summary改变状态) async def change_task_status(id:int,task:TaskChangeStatus): infoawait Task.filter(idid).update(**dict(task)) return { msg:ok, code:200, data:info }class TaskChangeStatus(BaseModel): status:intField( ..., title任务状态, description任务状态, example0 )只需要在basemodelzhong写要修改的字段即可这样就只能修改约定好的字段删task_router.delete(/delete/{id},summary删除任务) async def delete_task(id:int): infoawait Task.filter(idid).delete() return { msg:ok, code:200, data:info }删除只需要传要删除的id即可这里也是最简单的方法如果对于不存在的id也会返回删除成功实际上不存在的数据执行删除也没有意义因此加不加判断意义不大存在的直接删除。总结这里主要是对ORM 增删改查查询需先拼接 QuerySet 条件最后 await 避免列表报错新增用Task.create(**dict(模型))、修改用filter().update()、删除用filter().delete()局部修改可精简请求体模型仅保留允许更新的字段。其中对于新增修改等需要常用请求体传参的方法需要在basemodel中阐明相关字段是必须传...还是可穿可不穿None。当然方法不应该固化成一种有许多种方法实现尤其是新增还有很多实用的方法。