FastAPI+TortoiseORM 异步开发员工管理接口｜CRUD + 多条件分页 踩坑实录 FastAPITortoiseORM 异步开发员工管理接口CRUD 多条件分页 踩坑实录前言最近一直在练习 FastAPI TortoiseORM 异步后端开发搭建了一套简易的员工管理接口包含部门、员工、员工档案三组关联数据。表关系设计部门与员工一对多员工和员工档案一对一。所有接口实现基础 CRUD员工列表支持多条件模糊搜索 分页。写代码的时候图快踩了不少典型坑比如分层混乱、分页统计错误、异常处理不规范等。把完整代码和踩坑经验整理出来当作学习记录也给正在入门这套异步技术栈的小伙伴做参考。项目采用经典分层结构router路由层接收请求、参数转发、统一返回结果service业务层所有数据库操作、业务校验逻辑modelsTortoise ORM 数据表模型schemasPydantic 请求校验结构体个人习惯路由层尽量轻量化绝不直接写查询语句业务逻辑全部下沉到 Service。一、路由代码 day03_router.pyimportdatetimeimportmathfromfastapiimportAPIRouter,Queryfromtortoise.expressionsimportQfromapp.models.day03importDepartment,Employee,EmployeeProfilefromapp.schemas.day03importCreateDepartment,UpdateDepartment,CreateEmployee,UpdateEmployee,CreateEmployeeProfile,UpdateEmployeeProfilefromapp.services.day03importDay03Service day03_routerAPIRouter(prefix/day03,tags[day03])day03_router.get(/department)asyncdefdepartment():dataawaitDay03Service.department()return{msg:ok,code:200,data:data}day03_router.post(/create_department)asyncdefcreate_department(department:CreateDepartment):awaitDay03Service.create_department(department)return{msg:ok,code:200}day03_router.put(/update_department/{id})asyncdefupdate_department(id:int,department:UpdateDepartment):awaitDay03Service.update_department(id,department)return{msg:ok,code:200}day03_router.delete(/delete_department/{id})asyncdefdelete_department(id:int):awaitDay03Service.delete_department(id)return{msg:ok,code:200}day03_router.get(/employee)asyncdefemployee(name:strNone,department:strNone,status:strNone,page:intQuery(1,ge1,title页码,description页码),size:intQuery(1,ge1,le10,title每页条数,description每页条数)):data_list,page_infoawaitDay03Service.employee(name,department,status,page,size)return{msg:ok,code:200,data:{data_list:data_list,page_info:page_info}}day03_router.post(/create_employee)asyncdefcreate_employee(employee:CreateEmployee):awaitDay03Service.create_employee(employee)return{msg:ok,code:200}day03_router.put(/update_employee/{id})asyncdefupdate_employee(id:int,employee:UpdateEmployee):awaitDay03Service.update_employee(id,employee)return{msg:ok,code:200}day03_router.delete(/delete_employee/{id})asyncdefdelete_employee(id:int):awaitDay03Service.delete_employee(id)return{msg:ok,code:200}day03_router.get(/profile)asyncdefprofile():dataawaitDay03Service.profile()return{msg:ok,code:200,data:data}day03_router.get(/employee_profile/{id})asyncdefemployee_profile(id:int):infoawaitDay03Service.employee_profile(id)return{msg:ok,code:200,data:info}day03_router.post(/create_profile)asyncdefcreate_profile(employee_profile:CreateEmployeeProfile):awaitDay03Service.create_employee_profile(employee_profile)return{msg:ok,code:200}day03_router.put(/update_profile/{id})asyncdefupdate_profile(id:int,employee_profile:UpdateEmployeeProfile):awaitDay03Service.update_employee_profile(id,employee_profile)return{msg:ok,code:200}day03_router.delete(/delete_profile/{id})asyncdefdelete_profile(id:int):awaitDay03Service.delete_employee_profile(id)return{msg:ok,code:200}二、业务服务层 day03_service.py核心业务逻辑全部放在这里最初版本存在不少问题下文统一梳理坑点。importmathfromfastapiimportQueryfromtortoise.expressionsimportQfromapp.models.day03importDepartment,Employee,EmployeeProfilefromapp.schemas.day03importCreateDepartment,UpdateDepartment,CreateEmployee,UpdateEmployee,CreateEmployeeProfile,UpdateEmployeeProfileclassDay03Service:staticmethodasyncdefdepartment():dataawaitDepartment.all()returndatastaticmethodasyncdefcreate_department(department:CreateDepartment):dataawaitDepartment.get_or_none(namedepartment.name)ifdataisnotNone:raiseException(部门已存在)awaitDepartment.create(**dict(department))return200staticmethodasyncdefupdate_department(id:int,department:UpdateDepartment):dataawaitDepartment.get_or_none(idid)ifdataisNone:raiseException(部门不存在)department_dictdepartment.dict(exclude_unsetTrue)awaitDepartment.filter(idid).update(**department_dict)return200staticmethodasyncdefdelete_department(id:int):dataawaitDepartment.get_or_none(idid)ifdataisNone:raiseException(部门不存在)ifawaitEmployee.filter(departmentdata).exists():raiseException(部门下有员工无法删除)awaitDepartment.filter(idid).delete()return200staticmethodasyncdefemployee(name:strNone,department:strNone,status:strNone,page:intQuery(1,ge1,title页码,description页码),size:intQuery(1,ge1,le10,title每页条数,description每页条数)):offset(page-1)*size queryEmployee.all()ifname:queryquery.filter(Q(name__icontainsname))ifdepartment:queryquery.filter(departmentdepartment)ifstatus:queryquery.filter(statusstatus)dataawaitquery.prefetch_related(department).offset(offset).limit(size)data_list[]foriindata:data_info{id:i.id,name:i.name,emp_no:i.emp_no,gender:i.gender,age:i.age,phone:i.phone,email:i.email,hire_date:i.hire_date,salary:i.salary,department:i.department.name,status:i.status,created_at:i.created_at,updated_at:i.updated_at}data_list.append(data_info)page_info{page:page,size:size,total:awaitEmployee.all().count(),total_page:math.ceil(awaitEmployee.all().count()/size)}returndata_list,page_infostaticmethodasyncdefcreate_employee(employee:CreateEmployee):datadict(employee)data[department_id]data.pop(department)awaitEmployee.create(**dict(data))return200staticmethodasyncdefupdate_employee(id:int,employee:UpdateEmployee):dataawaitEmployee.get_or_none(idid)ifdataisNone:raiseException(员工不存在)employee_dictemployee.dict(exclude_unsetTrue)employee_dict[department_id]employee_dict.pop(department)awaitEmployee.filter(idid).update(**employee_dict)return200staticmethodasyncdefdelete_employee(id:int):dataawaitEmployee.get_or_none(idid)ifdataisNone:raiseException(员工不存在)awaitEmployee.filter(idid).delete()awaitEmployeeProfile.filter(employeeid).delete()return200staticmethodasyncdefprofile():dataawaitEmployeeProfile.all()returndatastaticmethodasyncdefemployee_profile(id:int):dataawaitEmployeeProfile.get_or_none(employeeid)ifdataisNone:raiseException(员工档案不存在)infoawaitEmployeeProfile.filter(employeeid)returninfostaticmethodasyncdefcreate_employee_profile(employee_profile:CreateEmployeeProfile):dataawaitEmployeeProfile.get_or_none(employeeemployee_profile.employee)ifdataisnotNone:raiseException(员工档案已存在)infodict(employee_profile)info[employee_id]info.pop(employee)awaitEmployeeProfile.create(**dict(info))return200staticmethodasyncdefupdate_employee_profile(id:int,employee_profile:UpdateEmployeeProfile):dataawaitEmployeeProfile.get_or_none(idid)ifdataisNone:raiseException(员工档案不存在)employee_profile_dictemployee_profile.dict(exclude_unsetTrue)awaitEmployeeProfile.filter(idid).update(**employee_profile_dict)return200staticmethodasyncdefdelete_employee_profile(id:int):dataawaitEmployeeProfile.get_or_none(idid)ifdataisNone:raiseException(员工档案不存在)awaitEmployeeProfile.filter(idid).delete()return200三、开发自测发现的问题重点写完基础功能进行接口测试陆续发现不少隐藏 bug这里记录下来。Query 依赖不能写在 Service 层最开始图省事直接把Query()定义在 service 方法参数上。后来才意识到Query是 FastAPI 路由专用依赖对象只能在接口函数使用。Service 属于通用业务层如果后续单元测试、被其他地方调用直接会报错。解决方案分页参数 page、size 仅保留在 routerservice 只接收普通 int 变量。分页统计总数逻辑错误原始代码统计总数写法totalawaitEmployee.all().count()这里有两个问题统计的是整张表所有员工没有拼接搜索条件用户筛选数据后分页总数不会变化前端页码错乱。连续两次执行 count ()发起两次 SQL 查询浪费数据库性能。解决方案count 调用基于拼接完条件的 query 对象await query.count()用变量接收一次结果复用。3. 直接抛出原生 Exception返回格式不统一代码里直接 raise Exception(“提示文字”)。FastAPI 捕获原生异常默认返回 500 错误页面无法和项目统一 JSON 返回格式。优化方向自定义业务异常类搭配全局异常处理器捕获后统一返回 {“code”:400,“msg”:“xxx”}。4. 删除员工手动清理档案存在数据不一致风险删除员工时先删员工记录再手动删除员工档案。如果两条 SQL 执行中途程序崩溃会出现员工删除成功、档案残留脏数据。两种优化方案任选其一使用in_transaction()事务包裹删除逻辑保证原子性ORM 模型外键定义 on_deleteONDELETE.CASCADE数据库层面自动级联删除。5. employee_profile 重复查询数据库dataawaitEmployeeProfile.get_or_none(employeeid)infoawaitEmployeeProfile.filter(employeeid)先查询一次判断数据是否存在紧接着再次查询同一条记录产生多余 IO。员工和档案是一对一关系可以简化查询逻辑。6. Pydantic v2 中 dict () 方法存在警告我本地环境使用 Pydantic v2.dict() 官方已经标记为过时运行时会输出警告。新项目建议统一替换为 .model_dump()。7. 外键字段转换代码重复新增、修改员工时需要手动将department重命名为department_id。多处重复相同逻辑后期可以封装公共方法简化代码。四、后续优化规划增加事务管理保证增删改操作的数据原子性封装通用分页工具类避免每个列表接口重复编写分页代码实现自定义业务异常 全局异常捕获使用 Pydantic 模型序列化 ORM 对象替代手动循环组装字典增加业务字段校验例如薪资不能为负数将重复的 “数据存在性校验” 抽取成公共工具函数。总结FastAPI 搭配 TortoiseORM 开发异步 CRUD 接口效率很高但是分层规范一定要守住。很多新手习惯直接在路由写数据库查询短期开发速度快后期迭代、维护成本极高。异步 ORM 和传统同步 ORM 写法上有不少区别需要留意避免不必要的重复查询。分页、联表查询、外键映射都是新手高频踩坑点。如果你也在学习这套技术栈可以拿这份代码当作基础模板进行改造。遇到问题欢迎评论区一起交流。