当前位置：首页 > news >正文

FastAPI 入门的后续以及Tortoise-ORM集成

news 2026/7/20 19:03:02

一、查询参数（Query Parameters）：

查询参数是URL中?后面的键值对组合，格式为key1=value1&key2=value2，用于对资源进行「筛选、分页、排序」等辅助操作。例如：

/items?skip=0&limit=10：skip（跳过条数）、limit（查询条数）是查询参数
/users?name=张三&age=20：name（姓名）、age（年龄）是查询参数
核心特点：
可选性：默认可省略，可设置默认值
辅助性：不用于标识唯一资源，仅用于过滤、分页等
灵活性：支持单个键对应多个值（如/items?tags=fruit&tags=cheap）

为什么需要Query类型注解？

基础的查询参数写法（如skip: int = 0）只能实现「类型校验+默认值」，但实际开发中需要更精细的控制：

分页参数limit必须≥1且≤50（范围校验）
搜索关键词q长度必须≤100（长度限制）
筛选标签tags支持多个值传入（多值参数）
接口文档需要显示查询参数的详细描述（元数据配置）
Query类型注解正是为解决这些问题而生，它是FastAPI提供的「查询参数高级配置工具」，与Path注解同源（均基于Pydantic），功能互补。

查询参数 vs 路径参数（核心区别）

什么是Query类型注解？

Query是FastAPI从fastapi模块导出的专用类，用于对查询参数进行「精细化配置」，功能与Path注解一致，仅适用场景不同。
核心特点：

兼容Python原生类型注解，支持更丰富的校验规则
配置自动同步到/docs接口文档，提升可读性
基于Pydantic实现，校验失败返回标准化422错误
支持多值参数、正则匹配等高级特性

Query注解最简示例

```python# ====================== Query类型注解基础示例 ======================fromfastapiimportFastAPI,Queryimportuvicorn app=FastAPI(title="Query注解教程",version="1.0.0")# Query注解：限制limit≥1且≤50，添加详细描述@app.get("/items/advanced/",summary="Query注解基础示例")defread_items_advanced(# 核心语法：参数名: 类型 = Query(默认值, 校验规则/元数据)skip:int=Query(0,ge=0,description="跳过条数，不能为负数"),limit:int=Query(10,ge=1,le=50,description="查询条数，1-50条")):""" Query注解分页接口 :param skip: 跳过条数（≥0） :param limit: 查询条数（1-50） :return: 分页结果 """fake_items=[{"item_id":i,"name":f"物品{i}"}foriinrange(skip,skip+limit)]return{"code":200,"skip":skip,"limit":limit,"data":fake_items}if__name__=="__main__":uvicorn.run("main:app",host="127.0.0.1",port=8000,reload=True)

### 核心校验规则参数 ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/4675a7b1a3f24f43ae342a020bbfca9a.png#pic_center) # 二、请求体与 Pydantic 模型 请求体解决的问题： 1.- 路径参数：只能传递简单值（ID、名称），且长度有限 2.- 查询参数：适合传递少量辅助数据，传递复杂数据（如用户注册信息、商品详情）时 URL 会冗长、不安全 优势： 数据容量大、格式灵活（支持 JSON / 表单 / 文件）、传输安全（配合 HTTPS） #### 三个核心参数类型的适用场景对比 ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/71dd17f6c9934ae88feaa15bb861da37.png#pic_center) ## 请求体通常与「非查询类」HTTP 方法配合使用（符合 RESTful 规范）： POST：创建资源（如用户注册、新增商品）→ 必用请求体 - PUT：全量更新资源（如修改商品所有信息）→ 必用请求体 - PATCH：部分更新资源（如修改商品价格）→ 常用请求体 - GET：查询资源 → 禁止使用请求体（不符合 HTTP 规范） ## 带字段校验的请求体模型 ```python ```python # ====================== Pydantic字段校验示例 ====================== from fastapi import FastAPI from pydantic import BaseModel, Field import uvicorn app = FastAPI(title="请求体字段校验教程", version="1.0.0") # 带字段校验的用户注册模型 class UserCreateWithValidate(BaseModel): """带字段校验的用户注册请求体模型""" # 用户名：3-20位，仅字母/数字/下划线，必填 username: str = Field( ..., # 必填字段 min_length=3, max_length=20, pattern=r"^[a-zA-Z0-9_]+$", title="用户名", description="3-20位，仅支持字母、数字、下划线", example="zhangsan_123" ) # 邮箱：符合邮箱格式，必填 email: str = Field( ..., pattern=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$", title="邮箱", description="请输入合法的邮箱地址", example="zs@test.com" ) # 密码：6-20位，必填 password: str = Field( ..., min_length=6, max_length=20, title="密码", description="6-20位字符，建议包含字母和数字", example="123456a" ) # 年龄：1-120岁，可选（默认None） age: int | None = Field( None, ge=1, le=120, title="年龄", description="1-120岁之间", example=25 ) # 带校验的注册接口 @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 "未填写" } } if __name__ == "__main__": uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

# 三、什么是 ORM？为什么要用 Tortoise-ORM？ ORM 全称是 Object-Relational Mapping（对象关系映射） 。它的核心思想是：用 Python 类来代表数据库中的表，用类的实例来代表表中的一行记录。 没有 ORM 时，你需要手写 SQL 语句来操作数据库。 ## ORM 的优势 - 面向对象：用 Python 代码替代 SQL 语句，更符合编程思维。 - 安全性：自动进行参数化查询，防止 SQL 注入攻击。 - 跨数据库：同一套代码可以无缝切换 SQLite、PostgreSQL、MySQL 等数据库。 - 关系管理：自动处理表与表之间的外键、多对多等关系。 - 可维护性：表结构集中定义在模型类中，修改和管理更方便。 ### 为什么选择 Tortoise-ORM？ 在 Python 异步 Web 开发中，传统的 ORM（如 SQLAlchemy 1.x 的同步模式）在执行数据库查询时会阻塞整个线程，这与 FastAPI 的异步非阻塞理念背道而驰，通常使用Tortoise-ORM或者SQLAlchemy 2.0。 ### 同步 ORM vs 异步 ORM 对比 ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/bfd81b2d5a564283972dc20636a50acf.png#pic_center) ## 环境搭建与 FastAPI 集成 ```python pip install tortoise-orm #MySQL 异步驱动（推荐 asyncmy） pip install asyncmy 或者 pip install aiomysql # 安装 Aerich 迁移工具（后面会用到） pip install aerich # 安装 FastAPI 和 Uvicorn pip install fastapi "uvicorn[standard]"

项目结构规划

数据库配置文件

配置中最重要的就是这一部分。

关系字段的 on_delete 策略详解

在定义 ForeignKeyField 或 OneToOneField 时，必须指定 on_delete 参数，它定义了当父表记录被删除时，子表关联记录的行为。这是保证数据一致性的重要一环

单表查询

先定义User模型 app/models/user.py，再导出user模型app/models/init.py，然后Aerich 数据库迁移，最后查询数据app/routers/user.py

示例：

fromdatetimeimportdatetime,date,timedeltafromfastapiimportAPIRouter,Queryfromtortoise.expressionsimportQfromapp.modelsimportTaskfromapp.schemas.day01importTaskCreateRequest task_router=APIRouter(prefix="/task",tags=["任务管理"],)@task_router.get("/all",summary="获取所有任务",description="获取所有任务")asyncdefgetAllTask(status:int|None=Query(None,description="0待办 1进行中 2已完成 3已取消，不传查全部"),keyword:str=Query("",description="任务标题模糊搜索关键词"),sort_priority:bool=Query(False,description="True=按优先级紧急→高→中→低排序，False=默认创建时间倒序")):query=Q()ifstatusisnotNone:query&=Q(status=status)ifkeyword.strip():query&=Q(title__icontains=keyword.strip())task_query=Task.filter(query)# 3. 优先级排序：紧急(3)→高(2)→中(1)→低(0)，降序ifsort_priority:task_query=task_query.order_by("-priority")else:# 默认按创建时间倒序task_query=task_query.order_by("-created_at")tasks=awaittask_query.all()# 通过任务状态查询tasks_list=[]fortaskintasks:tasks_list.append({"id":task.id,"title":task.title,"status":task.status,"priority":task.priority,"due_date":task.due_date,"created_at":task.created_at,})return{"code":1,"message":"success","data":tasks_list}

注意：一定要在在man.py中注册子路由
注释：上面示例写了查全部以及条件查询和排序，并在其中查询时做判断，没传就为空，或者为设置的默认值，传了直接查询。

单表增加

和查询过程一样，导包和名称不再展示，示例：

@task_router.post("/save",summary="保存任务",description="保存任务")asyncdefsave_task(task:TaskCreateRequest):task1=awaitTask.create(user_id=task.user_id,title=task.title,description=task.description,status=task.status,priority=task.priority,due_date=task.due_date)return{"code":1,"message":"保存成功","data":task1}

其中所要添加的字段我已在schemas中验证，会在最后展示它全部的代码

单表修改

示例：

@task_router.put("/update/{id}",summary="修改数据",description="修改数据")asyncdefupdate_task(id:int,task:TaskCreateRequest):task1=awaitTask.get_or_none(id=id)iftask1isNone:return{"code":0,"message":"任务不存在"}task_dict=task.dict(exclude_unset=True)awaitTask.filter(id=id).update(**task_dict)return{"code":1,"message":"修改成功"}

单表删除

示例;

@task_router.delete("/delete/{id}",summary="删除任务",description="删除任务")asyncdefdelete_task(id:int):task1=awaitTask.get_or_none(id=id)iftask1isNone:return{"code":0,"message":"任务不存在"}awaitTask.filter(id=id).delete()return{"code":1,"message":"删除成功"}

schemas的代码

frompydanticimportBaseModel,FieldclassTaskCreateRequest(BaseModel):user_id:int=Field(...,title="用户ID",description="用户ID",example=1)title:str=Field(...,title="任务标题",min_length=1,max_length=100,description="任务标题",example="学习FastAPI")description:str=Field(None,title="任务描述",min_length=1,max_length=1000,description="任务描述",example="学习FastAPI")status:int=Field(0,title="任务状态",description="任务状态",example=0)priority:int=Field(1,title="任务优先级",description="任务优先级",example=1)#截止时间不能早于当前时间due_date:str=Field(None,title="任务截止时间",description="任务截止时间",example="2026-07-20 00:00:00")classTaskUpdateRequest(BaseModel):title:str=Field(None,title="任务标题",min_length=1,max_length=100,description="任务标题",example="学习FastAPI")description:str=Field(None,title="任务描述",min_length=1,max_length=1000,description="任务描述",example="学习FastAPI")status:int=Field(None,title="任务状态",description="任务状态",example=0)priority:int=Field(None,title="任务优先级",description="任务优先级",example=1)due_date:str=Field(None,title="任务截止时间",description="任务截止时间",example="2026-07-20 00:00:00")completed_at:str=Field(None,title="任务完成时间",description="任务完成时间",example="2026-07-20 00:00:00")