FastAPI 学习日记 02：crud 和 api：搬运工把数据搬进搬出，门卫在前台开门迎客

大家好这是 FastAPI 学习日记第二篇。第 1 篇我们搭建好了项目全局架构吃透底层数据层models 数据表结构和schemas 接口协议校验。这一篇顺着请求链路拆解工程第二层核心crud 数据搬运层 api 路由接口层。建议先回看第一篇的完整请求链路图本篇所有知识点沿着链路逐层落地打通异步请求全流程。先给大家三句话速览本篇定位crud纯粹的数据搬运工负责数据库增删改查只处理数据不对外接收请求api前台门卫负责路由分发、参数接收、权限校验、结果组装返回AsyncSession贯穿两层的核心载体也是 FastAPI 异步架构、依赖注入设计的精髓一、先看 crud纯粹的数据搬运工crud 层是 FastAPI 工程中专门承载数据库操作的独立层级遵循一个业务模块对应一个 crud 文件。所有 SQL 查询、事务逻辑统一收敛在这里路由层只负责调用不手写查询语句实现业务与数据操作解耦。下面是分类模块完整可运行的 crud 代码python运行from typing import Sequence from sqlalchemy import select from sqlalchemy.ext.asyncio.session import AsyncSession from models.category import Category class CategoryCRUD: staticmethod async def get_all(db: AsyncSession, **kwargs) - Sequence[Category]: result await db.execute(select(Category).order_by(Category.id.desc())) return result.scalars().all() staticmethod async def create(db: AsyncSession, **kwargs) - Category: category Category(**kwargs) db.add(category) await db.commit() await db.refresh(category) return category category_crud CategoryCRUD()文件内实现两个基础方法全量查询、新增数据。代码看着简洁但每一行都是异步 SQLAlchemy2.0 的标准写法下面逐行拆解底层逻辑。1.1 解剖函数签名异步 crud 标准函数格式python运行async def create(db: AsyncSession, **kwargs) - Category:逐段拆解含义形成可复用模板表格片段作用详解async声明异步函数适配 FastAPI 异步架构调用时必须搭配awaitcreate语义化方法名直观表达业务动作db: AsyncSession异步数据库会话由外部依赖注入传入crud 不创建、不主动关闭会话**kwargs动态接收参数字典灵活适配新增、修改等可变传参场景- Category返回值类型注解标明返回 ORM 模型实例开启 IDE 智能提示与静态类型检测重点理解db是 crud 操作数据库唯一工具。crud 只负责使用会话执行操作会话生命周期管理交给 core 层彻底解耦。1.2 解剖查询链路execute → scalars → all 完整流程python运行result await db.execute(select(Category).order_by(Category.id.desc())) return result.scalars().all()这是单表查询标准写法分层拆解执行逻辑plaintextresult ← 接收数据库返回的原始 Result 对象并非直接是 ORM 对象列表 await ← 等待 I/O 操作期间主动让出 CPU不会阻塞服务 db.execute() ← 通过会话发送拼接完成的 SQL select(Category) ← 等价 SQLSELECT * FROM categories .order_by(Category.id.desc()) ← 追加排序等价 ORDER BY id DESC新手高频误区result 不是数据列表db.execute()返回Result 结果集对象内部数据默认封装为元组无法直接参与业务逻辑。 需要两层解析才能拿到纯净 ORM 对象.scalars()剥离外层元组外壳提取内部 ORM 模型实例.all()将迭代器转换成标准 Python 列表核心区别result.all()→[(Category实例,), (Category实例,)]元组嵌套结构result.scalars().all()→[Category实例, Category实例]纯净对象列表为什么必须使用 scalars ()单表查询场景下SQL 每行结果只对应一个 ORM 对象scalars 专门用于单列对象解析是单表查询最优方案。1.3 解剖新增链路add → commit → refresh 三步必懂python运行category Category(**kwargs) db.add(category) await db.commit() await db.refresh(category) return category异步新增固定流程每一步运行层级完全不同表格步骤执行动作运行层级核心说明实例化对象Category(**kwargs)Python 内存字典解包生成内存 ORM 对象无数据库交互加入会话db.add(category)会话缓冲区仅标记待新增不会发送 SQL延迟写入提交事务await db.commit()数据库真正执行 INSERT数据持久化入库回读刷新await db.refresh(category)数据库→内存同步数据库自增 ID、默认时间等自动生成字段三个新手必踩坑点一次性梳理清楚坑 1db.add () 不会写入数据库依靠 ORM 延迟写入机制add 只是把对象加入会话待办清单。多条操作可以最后统一 commit保证事务原子性要么全部成功要么全部回滚。坑 2commit 之后必须 refresh自增主键id、数据库默认时间server_defaultfunc.now()都是数据库侧生成的值。实例化对象时 Python 内存中不存在这些数据。commit 入库后必须 refresh 主动查询数据库同步数据到内存对象。呼应第一篇知识点使用 server_default 数据库默认值commit 后必须 refresh使用 Python 层 default 无需刷新。坑 3refresh 必须写在 commit 之后未执行 commit数据库不存在这条记录refresh 无法读取数据直接引发异常或者读取空值顺序不能颠倒。1.4 async/await 核心本质本篇最重要知识点很多开发者使用 FastAPI 只会照搬 async/await不理解底层原理这也是和 Django 同步开发最大的思维鸿沟。1、async 函数特性async def定义的异步函数直接调用不会执行代码只会生成协程对象必须搭配await函数逻辑才会启动运行。2、await 到底在等待什么await专门用于 I/O 阻塞场景数据库查询、网络请求、文件读写。 含义等待数据库完成 I/O 响应同时主动释放 CPU 资源处理其他用户请求不会阻塞整个服务。通俗场景对比Django 同步WSGI类似窗口排队一个请求阻塞数据库查询时后续所有请求全部等待FastAPI 异步ASGI类似餐厅点餐等待菜品的同时可以接待其他客人单进程支持高并发FastAPI 高性能核心不靠多线程、多进程依靠 I/O 等待时释放 CPU 资源。1.5 为什么一个请求对应独立 Session事务隔离单次请求对应独立事务请求结束事务生命周期终止避免多请求数据交叉污染非线程安全AsyncSession 不能并发共享全局共用会话会引发 SQL 异常、数据错乱1.6 全局实例化category_crud CategoryCRUD ()工程规范强制写法两个核心原因统一导入外部代码直接导入全局实例无需重复创建类对象代码简洁预留扩展后续如需改造实例方法外部调用代码无需改动低耦合补充全局实例无状态、不缓存任何数据多请求共用完全安全。二、再看 api前台门卫 路由分发api 层是项目唯一对外暴露入口直接接收前端请求。核心职责路由匹配、参数校验、依赖注入、调用 crud、封装响应。下面是分类模块标准路由代码python运行from typing import List from fastapi import APIRouter from api.dependencies import SessionDep from crud.category import category_crud from schemas.category import CategoryCreate, CategoryResp router APIRouter(prefix/categories, tags[v1 - 分类模块]) router.get(/, response_modelList[CategoryResp]) async def list_categories(db: SessionDep): 获取全部分类 return await category_crud.get_all(db) router.post(/, response_modelCategoryResp, status_code201) async def create_category(data: CategoryCreate, db: SessionDep): 新增分类 return await category_crud.create(db, **data.model_dump())可以清晰看到路由层没有任何 SQL、复杂业务逻辑只完成参数接收与 crud 调用做到极致解耦。2.1 依赖注入解密db 会话从哪里来crud 需要 AsyncSession但不会自己创建会话全部由 api 层依靠依赖注入自动供给。 会话源头定义在core/database.pypython运行async def get_db() - AsyncGenerator[AsyncSession, None]: async with async_session() as session: try: yield session finally: await session.close()yield 完整生命周期单次 HTTP 请求请求进入路由FastAPI 识别函数需要 db 依赖自动执行 get_db创建全新 AsyncSessionyield 将会话交给路由、crud 使用函数暂停等待业务逻辑执行完毕正常返回 / 抛出异常finally 代码强制执行关闭会话连接归还连接池为什么选择 yield不用 returnreturn 执行结束直接退出函数没有机会执行资源回收逻辑yield 可以交付资源、暂停执行等待业务代码结束后执行收尾操作完美实现「请求创建会话用完强制关闭异常自动兜底」。2.2 SessionDepAnnotated 优雅封装依赖为避免路由代码大量冗余工程在api/dependencies.py统一封装依赖别名python运行SessionDep Annotated[AsyncSession, Depends(get_db)]逐段拆解AsyncSession参数真实类型Depends(get_db)告知 FastAPI该参数通过执行 get_db 自动生成AnnotatedPython 原生类型增强工具不改变原有类型附加依赖规则新旧写法对比python运行# 老旧写法代码冗余 async def list_categories(db: AsyncSession Depends(get_db)): # 新版规范写法 async def list_categories(db: SessionDep):优势依赖统一集中维护路由代码整洁后续修改会话逻辑只改动一处不需要修改所有接口。2.3 dependencies.py 设计规范所有跨模块公共依赖统一抽离到 dependencies.pySessionDep数据库会话全局通用PaginationDep分页参数列表接口通用get_current_user登录鉴权需要权限的接口通用核心思想横向通用能力抽离业务路由只关注业务逻辑和 Django 抽离公共权限类、分页器思路一致。2.4 路由逐层拼接解决 404 报错核心问题新手 90% 接口 404根源是不理解路由前缀逐层累加机制。FastAPI 路由路径是分层拼接单文件内路径只是相对路径plaintext1、模块层category.py → prefix/categories 2、版本层v1/router.py → prefix/api/v1 3、项目层main.py → 全局挂载v1路由 最终完整访问路径/api/v1/categories/关键结论 路由文件中router.get(/)代表相对路径不是根路径/2.5 常见状态码坑点区分404/405/422表格状态码报错含义核心原因404路径不存在前缀漏配置、路径拼写错误、路由没有挂载405方法不允许路径存在但没有定义对应 HTTP 方法只写 GET未写 POST422参数校验失败字段缺失、类型错误、长度超限schemas 在校验阶段直接拦截2.6 新增模块工程规范新增任意业务模块用户、商品、订单必须同步更新 4 个文件否则直接导入报错models/__init__.py导出新的数据表模型schemas/__init__.py导出新增、响应模型crud/__init__.py导出 crud 全局实例api/v1/router.py挂载新模块路由三、本篇学习小结再次串联完整请求链路形成知识闭环前端请求 → api 路由分发 依赖注入获取 session → crud 执行数据库操作 → schemas 序列化返回 JSON本篇核心知识点汇总crud 层定位纯数据搬运掌握execute→scalars→all查询流程、add→commit→refresh新增流程异步核心await 等待 I/O、让出 CPU是 FastAPI 高并发本质单请求单会话保证事务安全依赖注入yield 实现资源自动创建与回收Annotated 统一封装依赖代码解耦路由机制前缀逐层拼接分清 404/405/422 报错场景新增模块同步更新对应__init__.py下一篇预告下一篇进入整个项目最核心的地基 ——core 核心配置层 深度拆解engine 连接池原理、sessionmaker 会话工厂、Base 元数据机制、env 环境变量优先级、get_db 完整底层生命周期彻底吃透 FastAPI 异步工程的底层根基。这是我从 Django 转型 FastAPI 工程化重构的真实踩坑总结。如有错误或者理解不到位的地方欢迎评论区指正觉得有帮助可以点赞收藏下一篇见