第一部分Tortoise-ORM集成ORM 的概念ORM 全称是 Object-Relational Mapping对象关系映射 。它的核心思想是用 Python 类来代表数据库中的表用类的实例来代表表中的一行记录同步 ORM vs 异步 ORM 对比如代码所示# 同步写法Django ORMuserUser.objects.get(id1)postsPost.objects.filter(authoruser)# 异步写法Tortoise-ORM—— 注意 await 关键字userawaitUser.get(id1)postsawaitPost.filter(authoruser)第二部分环境搭建与 FastAPI 集成基于 MySQL# 安装 Tortoise-ORM 核心库pip install tortoise-orm#MySQL 异步驱动推荐 asyncmypip install asyncmy 或者 pip install aiomysql# 安装 Aerich 迁移工具后面会用到pip install aerich# 安装 FastAPI 和 Uvicornpip install fastapiuvicorn[standard]项目结构规划数据库配置文件app/config.pypythonapp/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.pyfromfastapiimportFastAPI,Query,HTTPExceptionfrompydanticimportField,BaseModelfromstarlette.requestsimportRequestfromstarlette.responsesimportJSONResponsefromtortoiseimportTortoisefromtortoise.contrib.fastapiimportregister_tortoisefromapp.configimportTORTOISE_ORMfromapp.routers.userimportuser_router appFastAPI()register_tortoise(appapp,configTORTOISE_ORM,generate_schemasTrue,add_exception_handlersTrue)第三部分完整项目流程Model模型搭建fromtortoiseimportmodels,fieldsclassUser(models.Model):idfields.BigIntField(pkTrue)usernamefields.CharField(max_length50,verbose_name用户名)passwordfields.CharField(max_length50,verbose_name密码)classMeta:tablet_usertable_description用户表classTask(models.Model):idfields.BigIntField(pkTrue)user_idfields.BigIntField(verbose_name所属用户)titlefields.CharField(max_length50,verbose_name任务标题)descriptionfields.TextField(nullTrue,verbose_name任务描述)statusfields.SmallIntField(default0,verbose_name任务状态)priorityfields.SmallIntField(default1,verbose_name任务优先级)due_datefields.DatetimeField(nullTrue,verbose_name任务截止日期)completed_atfields.DatetimeField(nullTrue,verbose_name完成时间)created_atfields.DatetimeField(auto_now_addTrue)updated_atfields.DatetimeField(auto_nowTrue)classMeta:tablet_tasktable_description任务表查询数据routers:fromdatetimeimportdatetimefromtypingimportListfromfastapiimportAPIRouter,Queryfromapp.models.bookimportUser,Taskfromapp.schemas.bookimportUserLoginRequest,TaskSchema,TaskResponse,TaskList,TaskCreate,TaskUpdate,TaskStatus user_routerAPIRouter(prefix/user,tags[用户])user_router.post(/login,summary登录,description登录)asyncdeflogin(user_login:UserLoginRequest):userawaitUser.filter(usernameuser_login.username).first()ifuserisNone:return{code:0,message:用户不存在}ifuser.password!user_login.password:return{code:0,message:密码错误}return{code:1,message:登陆成功}task_routerAPIRouter(prefix/task,tags[任务])task_router.get(/list,response_modelList[TaskResponse],summary任务列表筛选)asyncdeftask_list(query:TaskListQuery()):infoTask.filter(user_id1)ifquery.statusandquery.status!all:infoinfo.filter(statusint(query.status))ifquery.keyword:infoinfo.filter(title__containsquery.keyword)ifquery.sort_priorityasc:infoinfo.order_by(-priority)else:infoinfo.order_by(priority)returnawaitinfotask_router.post(/add,response_modelTaskResponse,summary新增任务)asyncdeftask_add(data:TaskCreate):taskawaitTask.create(**data.model_dump())returntasktask_router.delete(/delete/{id},summary删除任务,description删除任务)asyncdeftask_delete(id:int):awaitTask.filter(idid).delete()return{code:1,message:删除成功}task_router.put(/update/{id},summary更新用户,description更新用户)asyncdefupdate_task(task:TaskUpdate,id:int):print(dict(task))task1awaitTask.get_or_none(idid)iftask1isNone:return{code:0,message:task not found}task_dicttask.model_dump(exclude_unsetTrue)awaitTask.filter(idid).update(**task_dict)return{code:1,message:update success}task_router.put(/{id}/status/,summary更新任务状态,description快速切换任务状态自动处理完成时间)asyncdefupdate_task_status(body:TaskStatus,id:int):task1awaitTask.get_or_none(idid)iftask1isNone:return{code:0,message:task not found}update_databody.model_dump(exclude_unsetTrue)new_statusupdate_data.get(status)ifnew_status2:update_data[completed_at]datetime.now()else:update_data[completed_at]NoneawaitTask.filter(idid).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_routerAPIRouter(prefix/user,tags[用户])user_router.post(/login,summary登录,description登录)asyncdeflogin(user_login:UserLoginRequest):userawaitUser.filter(usernameuser_login.username).first()ifuserisNone:return{code:0,message:用户不存在}ifuser.password!user_login.password:return{code:0,message:密码错误}return{code:1,message:登陆成功}task_routerAPIRouter(prefix/task,tags[任务])task_router.get(/list,response_modelList[TaskResponse],summary任务列表筛选)asyncdeftask_list(query:TaskListQuery()):infoTask.filter(user_id1)ifquery.statusandquery.status!all:infoinfo.filter(statusint(query.status))ifquery.keyword:infoinfo.filter(title__containsquery.keyword)ifquery.sort_priorityasc:infoinfo.order_by(-priority)else:infoinfo.order_by(priority)returnawaitinfotask_router.post(/add,response_modelTaskResponse,summary新增任务)asyncdeftask_add(data:TaskCreate):taskawaitTask.create(**data.model_dump())returntasktask_router.delete(/delete/{id},summary删除任务,description删除任务)asyncdeftask_delete(id:int):awaitTask.filter(idid).delete()return{code:1,message:删除成功}task_router.put(/update/{id},summary更新用户,description更新用户)asyncdefupdate_task(task:TaskUpdate,id:int):print(dict(task))task1awaitTask.get_or_none(idid)iftask1isNone:return{code:0,message:task not found}task_dicttask.model_dump(exclude_unsetTrue)awaitTask.filter(idid).update(**task_dict)return{code:1,message:update success}task_router.put(/{id}/status/,summary更新任务状态,description快速切换任务状态自动处理完成时间)asyncdefupdate_task_status(body:TaskStatus,id:int):task1awaitTask.get_or_none(idid)iftask1isNone:return{code:0,message:task not found}update_databody.model_dump(exclude_unsetTrue)new_statusupdate_data.get(status)ifnew_status2:update_data[completed_at]datetime.now()else:update_data[completed_at]NoneawaitTask.filter(idid).update(**update_data)return{code:1,message:status update success}这是一个任务表与用户表格的接口效果实现如图下面是用户表的登录接口效果展示下面是任务表格的查询接口展示下面是给任务表格添加数据的效果下面是对任务表格数据修改的展示下面是对任务表格数据删除的展示下面是对任务表格状态的修改的展示以上也就是用orm对数据的增删改查的功能展示他的用法跟django的语法类似