第一部分：Tortoise-ORM集成

ORM 的概念

ORM 全称是 Object-Relational Mapping（对象关系映射） 。它的核心思想是：用 Python 类来代表数据库中的表，用类的实例来代表表中的一行记录
同步 ORM vs 异步 ORM 对比如代码所示：

# 同步写法（Django ORM）user=User.objects.get(id=1)posts=Post.objects.filter(author=user)# 异步写法（Tortoise-ORM）—— 注意 await 关键字！user=awaitUser.get(id=1)posts=awaitPost.filter(author=user)

第二部分：环境搭建与 FastAPI 集成（基于 MySQL）

# 安装 Tortoise-ORM 核心库pip install tortoise-orm#MySQL 异步驱动（推荐 asyncmy）pip install asyncmy 或者 pip install aiomysql# 安装 Aerich 迁移工具（后面会用到）pip install aerich# 安装 FastAPI 和 Uvicornpip install fastapi"uvicorn[standard]"

项目结构规划：

数据库配置文件（app/config.py）：```python

app/config.py

“”"
数据库配置文件
这个文件定义了 Tortoise-ORM 连接 MySQL 数据库所需的所有配置信息
“”"

TORTOISE_ORM 是 Tortoise-ORM 规定的配置字典变量名

后面用 register_tortoise 或 Aerich 时都会引用这个字典

TORTOISE_ORM = {
# 1. 连接配置 —— 定义数据库连接信息
“connections”: {
# “default” 是默认连接的名字，必须有一个 default
“default”: {
# engine：指定数据库后端引擎，MySQL 使用 tortoise.backends.mysql
“engine”: “tortoise.backends.mysql”,

# credentials：数据库连接凭证，包含主机、端口、用户名、密码等 "credentials": { "host": "127.0.0.1", # MySQL 服务器地址 "port": 3306, # MySQL 端口，默认 3306 "user": "root", # 数据库用户名 "password": "123456", # 数据库密码（请根据实际情况修改） "database": "fastapi_db1", # 数据库名称 "minsize": 1, # 连接池最小连接数 "maxsize": 5, # 连接池最大连接数 "charset": "utf8mb4", # 字符集，支持 emoji "echo": True # 是否打印 SQL 语句（开发环境建议开启） } } }, # 2. 应用配置 —— 指定模型所在的模块 "apps": { # "models" 是应用的名字，可以自定义，但 Aerich 需要使用这个名字 "models": { # models 列表：指定包含 Tortoise 模型类的 Python 模块路径 # "aerich.models" 是 Aerich 的内置模型，用于记录迁移历史，必须包含 "models": ["app.models", "aerich.models"], # default_connection：指定这个应用使用哪个数据库连接 "default_connection": "default", } }, # 3. 时区配置 "use_tz": False, # 是否使用时区 "timezone": "Asia/Shanghai" # 时区设置

}
``
FastAPI 主应用（app/main.py）：

fromfastapiimportFastAPI,Query,HTTPExceptionfrompydanticimportField,BaseModelfromstarlette.requestsimportRequestfromstarlette.responsesimportJSONResponsefromtortoiseimportTortoisefromtortoise.contrib.fastapiimportregister_tortoisefromapp.configimportTORTOISE_ORMfromapp.routers.userimportuser_router app=FastAPI()register_tortoise(app=app,config=TORTOISE_ORM,generate_schemas=True,add_exception_handlers=True)

第三部分：完整项目流程：
Model模型搭建：

fromtortoiseimportmodels,fieldsclassUser(models.Model):id=fields.BigIntField(pk=True)username=fields.CharField(max_length=50,verbose_name="用户名")password=fields.CharField(max_length=50,verbose_name="密码")classMeta:table="t_user"table_description="用户表"classTask(models.Model):id=fields.BigIntField(pk=True)user_id=fields.BigIntField(verbose_name="所属用户")title=fields.CharField(max_length=50,verbose_name="任务标题")description=fields.TextField(null=True,verbose_name="任务描述")status=fields.SmallIntField(default=0,verbose_name="任务状态")priority=fields.SmallIntField(default=1,verbose_name="任务优先级")due_date=fields.DatetimeField(null=True,verbose_name="任务截止日期")completed_at=fields.DatetimeField(null=True,verbose_name="完成时间")created_at=fields.DatetimeField(auto_now_add=True)updated_at=fields.DatetimeField(auto_now=True)classMeta:table="t_task"table_description="任务表"

查询数据routers:

fromdatetimeimportdatetimefromtypingimportListfromfastapiimportAPIRouter,Queryfromapp.models.bookimportUser,Taskfromapp.schemas.bookimportUserLoginRequest,TaskSchema,TaskResponse,TaskList,TaskCreate,TaskUpdate,TaskStatus user_router=APIRouter(prefix="/user",tags=["用户"])@user_router.post("/login",summary="登录",description="登录")asyncdeflogin(user_login:UserLoginRequest):user=awaitUser.filter(username=user_login.username).first()ifuserisNone:return{"code":0,"message":"用户不存在"}ifuser.password!=user_login.password:return{"code":0,"message":"密码错误"}return{"code":1,"message":"登陆成功"}task_router=APIRouter(prefix="/task",tags=["任务"])@task_router.get("/list",response_model=List[TaskResponse],summary="任务列表筛选")asyncdeftask_list(query:TaskList=Query()):info=Task.filter(user_id=1)ifquery.statusandquery.status!="all":info=info.filter(status=int(query.status))ifquery.keyword:info=info.filter(title__contains=query.keyword)ifquery.sort_priority=="asc":info=info.order_by("-priority")else:info=info.order_by("priority")returnawaitinfo@task_router.post("/add",response_model=TaskResponse,summary="新增任务")asyncdeftask_add(data:TaskCreate):task=awaitTask.create(**data.model_dump())returntask@task_router.delete("/delete/{id}",summary="删除任务",description="删除任务")asyncdeftask_delete(id:int):awaitTask.filter(id=id).delete()return{"code":1,"message":"删除成功"}@task_router.put("/update/{id}",summary="更新用户",description="更新用户")asyncdefupdate_task(task:TaskUpdate,id:int):print(dict(task))task1=awaitTask.get_or_none(id=id)iftask1isNone:return{"code":0,"message":"task not found"}task_dict=task.model_dump(exclude_unset=True)awaitTask.filter(id=id).update(**task_dict)return{"code":1,"message":"update success"}@task_router.put("/{id}/status/",summary="更新任务状态",description="快速切换任务状态，自动处理完成时间")asyncdefupdate_task_status(body:TaskStatus,id:int):task1=awaitTask.get_or_none(id=id)iftask1isNone:return{"code":0,"message":"task not found"}update_data=body.model_dump(exclude_unset=True)new_status=update_data.get("status")ifnew_status==2:update_data["completed_at"]=datetime.now()else:update_data["completed_at"]=NoneawaitTask.filter(id=id).update(**update_data)return{"code":1,"message":"status update success"}

schemas:

fromdatetimeimportdatetimefromtypingimportListfromfastapiimportAPIRouter,Queryfromapp.models.bookimportUser,Taskfromapp.schemas.bookimportUserLoginRequest,TaskSchema,TaskResponse,TaskList,TaskCreate,TaskUpdate,TaskStatus user_router=APIRouter(prefix="/user",tags=["用户"])@user_router.post("/login",summary="登录",description="登录")asyncdeflogin(user_login:UserLoginRequest):user=awaitUser.filter(username=user_login.username).first()ifuserisNone:return{"code":0,"message":"用户不存在"}ifuser.password!=user_login.password:return{"code":0,"message":"密码错误"}return{"code":1,"message":"登陆成功"}task_router=APIRouter(prefix="/task",tags=["任务"])@task_router.get("/list",response_model=List[TaskResponse],summary="任务列表筛选")asyncdeftask_list(query:TaskList=Query()):info=Task.filter(user_id=1)ifquery.statusandquery.status!="all":info=info.filter(status=int(query.status))ifquery.keyword:info=info.filter(title__contains=query.keyword)ifquery.sort_priority=="asc":info=info.order_by("-priority")else:info=info.order_by("priority")returnawaitinfo@task_router.post("/add",response_model=TaskResponse,summary="新增任务")asyncdeftask_add(data:TaskCreate):task=awaitTask.create(**data.model_dump())returntask@task_router.delete("/delete/{id}",summary="删除任务",description="删除任务")asyncdeftask_delete(id:int):awaitTask.filter(id=id).delete()return{"code":1,"message":"删除成功"}@task_router.put("/update/{id}",summary="更新用户",description="更新用户")asyncdefupdate_task(task:TaskUpdate,id:int):print(dict(task))task1=awaitTask.get_or_none(id=id)iftask1isNone:return{"code":0,"message":"task not found"}task_dict=task.model_dump(exclude_unset=True)awaitTask.filter(id=id).update(**task_dict)return{"code":1,"message":"update success"}@task_router.put("/{id}/status/",summary="更新任务状态",description="快速切换任务状态，自动处理完成时间")asyncdefupdate_task_status(body:TaskStatus,id:int):task1=awaitTask.get_or_none(id=id)iftask1isNone:return{"code":0,"message":"task not found"}update_data=body.model_dump(exclude_unset=True)new_status=update_data.get("status")ifnew_status==2:update_data["completed_at"]=datetime.now()else:update_data["completed_at"]=NoneawaitTask.filter(id=id).update(**update_data)return{"code":1,"message":"status update success"}

这是一个任务表与用户表格的接口，效果实现如图：

1. 下面是用户表的登录接口效果展示：
2. 下面是任务表格的查询接口展示：
   
3. 下面是给任务表格添加数据的效果：
4. 下面是对任务表格数据修改的展示：
5. 下面是对任务表格数据删除的展示：
6. 下面是对任务表格状态的修改的展示：
   以上也就是用orm对数据的增删改查的功能展示，他的用法跟django的语法类似