


让 AI 编写 FastAPI 服务,并在审查中找出常见错误
在让 AI 编写或审查 FastAPI 服务时使用。规定最低版本、异步与同步路由的选择、Pydantic 模型与依赖注入的写法、数据库与测试的做法,并用表格列出审查代码差异时要找出的常见错误及改法。
要使用这个AI规则
## 设定 Python、FastAPI 及各个库的最低版本
将每个依赖锁定为下表版本或更新版本;本文的所有指导都基于这些基线。使用更旧的版本会悄无声息地破坏后续章节中的惯用写法——异步 API、`Annotated` 依赖注入以及 v2 序列化。
| 依赖 | 最低版本 | 设定该下限的原因 |
|---|---|---|
| Python | 3.11 | `StrEnum`,以及 `X \| Y` 联合类型运算符 |
| FastAPI | 0.115 | `Annotated[T, Depends(...)]` 是惯用写法 |
| Pydantic | 2.7 | v1 API(`json_encoders`、`.dict()`)已被移除 |
| pydantic-settings | 2.4 | 自 Pydantic v2 起独立为单独的包 |
| SQLAlchemy | 2.0 | 异步 API(`AsyncSession`、`async_sessionmaker`) |
| Alembic | 1.13 | 支持异步的迁移功能 |
| httpx | 0.27 | 用于进程内测试的 `ASGITransport` |
| PyJWT | 2.9 | 持续维护中;替代 `python-jose` |
| ruff | 0.6 | 一个工具搞定代码检查、格式化和导入排序 |
## 按领域而不是文件类型划分项目文件夹
按领域(限界上下文)而非文件类型来组织目录结构。为每个领域设置独立的包,并保留一个精简的全局层来处理共享关注点。
- 每个领域包内包含:`router.py`(接口端点)、`schemas.py`(Pydantic 模型)、`models.py`(ORM 模型)、`service.py`(业务逻辑)、`dependencies.py`、`config.py`(领域范围的配置)、`constants.py`、`exceptions.py`、`utils.py`。
- 全局顶层目录:`config.py`(全局配置)、`models.py`(共享基类)、`exceptions.py`、`database.py`(异步引擎与会话工厂)、`main.py`(应用与生命周期管理)。
- 跨领域导入:始终显式导入具体模块并为其取别名(例如,将某领域的 `service` 导入为 `auth_service`)。绝不使用通配符导入,也绝不深入内部路径进行引用——这两种做法都会造成领域间耦合并阻碍重构。
## 区分使用 async def 与 def 路由,避免应用卡住
根据路由实际执行的工作来选择路由风格;选错了要么会冻结事件循环,要么会浪费线程。
| 路由执行的操作 | 应使用 |
|---|---|
| 可等待的非阻塞 I/O | `async def` |
| 阻塞式 I/O,且不存在异步客户端 | `def`(同步,在线程池中运行) |
| 两者混合 | `async def` 搭配 `run_in_threadpool` 处理阻塞调用 |
| 耗时超过约 50 毫秒的 CPU 密集型工作 | 卸载到工作进程(Celery / Arq / RQ) |
绝不要在 `async def` 内部放入阻塞调用——比如 `time.sleep`、`requests`、同步数据库驱动或普通的 `open()`——这会冻结该工作进程上的每一个请求。同步的 `def` 路由只会阻塞一个线程池工作线程,因此它是承载不可避免的阻塞调用的安全场所。在 `async def` 内部,应使用可等待的等价方案(`asyncio.sleep`、`aiofiles`、异步驱动),或者用 `run_in_threadpool` 包装同步库。注意事项:默认线程池大约只有 40 个线程,占满它会拖慢每一个同步路由,而且线程的开销比协程更大——没有阻塞方面的理由就不要选择同步路由。
## 用 Pydantic 校验输入,并按领域拆分配置
- 优先使用内置校验器和类型,而非手写检查逻辑:`Field(min_length=, max_length=, pattern=, ge=)`、`EmailStr`、`AnyUrl`,以及用于封闭集合的 `StrEnum`。
- 不要将约束条件与相互矛盾的默认值搭配使用。`Field(ge=18, default=None)` 就是一个 bug:应明确该字段是必填(`Field(ge=18)`)还是可选(`int | None = Field(default=None, ge=18)`)。
- 使用现代化的序列化方式。`json_encoders` 在 Pydantic v2 中已被弃用——针对单个字段的规则应改用 `@field_serializer`,或者用 `PlainSerializer` 为类型添加注解。
- 按领域拆分配置。`pydantic-settings` 是独立的包;应为每个领域分配一个带有专属 `env_prefix` 的 `BaseSettings` 子类,而不是让所有领域共用一个全局配置对象。
## 用 Annotated 写依赖注入,并让依赖检查数据
- 使用 `Annotated[T, Depends(fn)]` 进行注入,这是自 FastAPI 0.95 起的惯用写法。避免使用默认参数形式(`param: T = Depends(fn)`);这是遗留写法,且存在默认值方面的隐患。
- 让依赖项承担校验职责,而不只是注入。一个依赖项可以按 id 加载资源,在资源缺失时抛出领域级的 404 错误,并返回已校验的对象,使路由接收到的是已经检查过的内容。
- 通过依赖链来复用逻辑。一个依赖项可以依赖于其他依赖项——例如,一个所有权检查依赖项可以同时依赖于一个按 id 加载的依赖项和一个已解析令牌的依赖项。
- 规则:依赖项在单次请求内会被缓存,因此同一个 `Depends(x)` 在一次请求中被多次使用也只会执行一次。优先使用 `async def` 依赖项;同步依赖项会在线程池中运行,对于仅涉及少量 CPU 运算的检查来说是不必要的开销。要在多个接口端点间复用同一个依赖项,只需为共享的路径变量使用相同的名称。
## 编写 JWT 身份验证与异步数据库访问
- JWT:使用 `PyJWT`(`import jwt`),而非已停止维护的 `python-jose`。在针对 `InvalidTokenError` 的保护逻辑中执行解码,并抛出领域级的凭证错误,而不是让原始异常直接抛出。
- 会话:优先使用 SQLAlchemy 2.0 的异步 API(`create_async_engine`、`async_sessionmaker`、`AsyncSession`),并通过 `get_db` 这个异步生成器依赖项将会话传递给路由。`encode/databases` 已进入维护模式——新项目不要选用它。
- 命名:使用小写蛇形命名(`lower_case_snake`);表名使用单数形式(`post`、`user`、`post_like`);用共同的前缀对相关表进行分组(`payment_account`、`payment_bill`);日期时间字段以 `_at` 结尾、日期字段以 `_date` 结尾;同一个外键列在所有出现的地方都使用相同的名称。
- 在 `MetaData` 上设置一套项目全局统一的索引与约束命名规范,使生成的名称保持可预测。
- SQL 优先、Pydantic 其次:在 SQL 中完成连接查询、聚合运算和 JSON 结构整形,仅在响应校验阶段才将数据填充进 Pydantic,而不是用它来做数据转换。
## 在后台任务与 Celery 等任务队列之间做选择
根据任务的开销和可靠性需求来选择合适的工具。
| 适合使用进程内后台任务的情形 | 适合使用真正的任务队列(Celery / Arq / RQ)的情形 |
|---|---|
| 任务能在一秒内完成 | 任务需要运行数秒到数分钟 |
| 任务悄悄丢失也可以接受 | 需要重试、死信队列或可观测性 |
| 任务是进程内操作(如发送邮件、记录一行日志) | 任务是 CPU 密集型的,或需要独立的资源池 |
| 不需要调度 | 需要 cron 定时、预计执行时间(ETA)或限流 |
进程内后台任务会在响应发送之后、在同一个工作进程中运行。如果该工作进程崩溃,任务就会丢失且不会重试,因此绝不要把它们用于任何需要呼叫值班工程师处理的重要事务上。
## 用 httpx 和真实数据库做测试
- 从一开始就使用异步 HTTP 客户端:通过 `ASGITransport(app=app)` 搭配 `httpx.AsyncClient` 在进程内驱动应用。不要使用 `async_asgi_testclient`——它已经停止维护。
- 使用框架内置的 `app.dependency_overrides[dep] = fake` 来覆盖依赖项,而不是对内部实现进行猴子补丁(monkeypatch),并在每次测试后清除这些覆盖设置。
- 集成测试应使用真实数据库(临时 schema 或一次性容器),只覆盖身份验证和外部服务。对数据库做 mock 会掩盖 mock 与生产环境之间的差异,这种差异最终会在生产环境中爆发。
## 让数据库迁移可回滚,并为每个 API 写好文档
- 迁移:确保每次迁移都是静态且可回滚的,从异步模板生成迁移文件,并使用带时间戳的描述性文件名(日期前缀加上一个 slug),使历史记录能够按顺序阅读。
- API 文档:当前环境不在允许列表中时,通过将 OpenAPI URL 设为 `None` 来隐藏交互式文档,避免其在非指定环境中暴露。为每个接口端点提供完整文档——响应模型、状态码、摘要、描述、标签,以及针对每种错误状态各自的响应条目。
- 代码检查:运行格式化工具和代码检查工具(例如先执行 `ruff check --fix`,再执行 `ruff format`),并将它们接入 pre-commit 钩子或 CI 流程。一个现代化工具就能取代原本各自独立的格式化工具、导入排序工具和死代码检测工具。
## 在 FastAPI 代码差异中找出 AI 的常见错误并修正
在审查代码差异时,应标记出以下这些代理常犯的错误;每一行都对应一种真实的失败模式及其修复方法。
| 反模式 | 为什么是错的 | 修复方法 |
|---|---|---|
| 在 `async def` 内使用 `requests.get(...)` | 会阻塞事件循环 | 改用 `httpx.AsyncClient`,或 `await run_in_threadpool(...)` |
| 在 `async def` 中使用 `time.sleep` / `open()` / 同步数据库驱动 | 会阻塞事件循环 | 改用对应的异步方案(`asyncio.sleep`、`aiofiles`、异步驱动) |
| 使用 `from jose import jwt` | 该库已停止维护 | 改用 `import jwt`(PyJWT) |
| 使用 `async_asgi_testclient` | 已停止维护 | 改用 `httpx.AsyncClient` 搭配 `ASGITransport` |
| 使用 `ConfigDict(json_encoders=...)` | 在 Pydantic v2 中已被弃用 | 改用 `@field_serializer` 或 `PlainSerializer` |
| 使用 `Field(ge=18, default=None)` | 约束条件与默认值相矛盾 | 明确选择必填或可选,二者不可兼得 |
| 使用默认参数形式的 `Depends(...)` | 存在遗留的默认值隐患 | 改用 `Annotated[T, Depends(...)]` |
| 在路由主体外捕获裸露的 `Exception` | 会掩盖 bug,把本应返回的 500 错误悄悄变成 200 | 应捕获具体的异常类,并显式抛出 HTTP 错误 |
| 对需要告警呼叫的重要工作使用后台任务 | 没有重试机制,会随工作进程一起消失 | 改用 Celery / Arq / RQ |
| 在 `async def` 中使用同步 ORM 会话 | 会阻塞事件循环,还可能导致连接池死锁 | 改用 `AsyncSession` |
| 返回模型的同时又将 `response_model` 设为同一模型 | 模型会被构建两次 | 改为返回字典或行数据,或者去掉 `response_model` |
| 深层跨领域导入 | 造成紧耦合 | 应导入领域模块并为其取别名 |
| 整个应用共用一个 `BaseSettings` | 每个领域都会读取所有变量 | 每个领域各自使用一个 `BaseSettings` |
| 在集成测试中对数据库做 mock | 会造成 mock 与生产环境之间的差异 | 应使用真实数据库,并只对身份验证和外部服务做依赖覆盖 |
