ByteNoteByteNote

字节笔记本

2026年8月29日

FastAPI 生产实践:按领域拆项目、别把事件循环堵死

API中转
¥120

很多人把 FastAPI 官方教程写完就上线了,项目一长大就开始乱:路由、模型、服务全堆在一层文件夹里,async 路由里还塞着 time.sleep。GitHub 上这份 fastapi-best-practices(约 1.8 万 star)是几个创业团队踩坑之后整理出来的约定,偏实战,不是教材复述。

下面挑几条最容易立刻用上的。

按领域拆,不要按文件类型拆

常见教程会搞 routers/models/crud/。微服务或小项目还行,领域一多就变成「改一个功能要横跳五个目录」。

更稳的做法是按业务域分包,每个域自己带齐文件:

text
src/
  auth/
    router.py
    schemas.py      # Pydantic
    models.py       # ORM
    service.py
    dependencies.py
    exceptions.py
    ...
  posts/
    router.py
    schemas.py
    models.py
    service.py
    ...
  config.py
  database.py
  main.py

跨域引用时写清楚模块名,别来个模糊的 from service import ...

python
from src.auth import constants as auth_constants
from src.notifications import service as notification_service

一眼能看出依赖从哪来,后续拆服务也少心智负担。

async 路由别堵事件循环

FastAPI 对 sync / async 路由的处理不一样:

  • def 路由会丢进线程池,阻塞 I/O 不至于卡死整站
  • async def 路由直接在事件循环上跑,你若在里面写 time.sleep 或同步 SDK,全站一起等

典型对比:

python
@router.get("/terrible-ping")
async def terrible_ping():
    time.sleep(10)  # 堵住事件循环
    return {"pong": True}

@router.get("/good-ping")
def good_ping():
    time.sleep(10)  # 在线程池里堵,主循环还能接别的请求
    return {"pong": True}

@router.get("/perfect-ping")
async def perfect_ping():
    await asyncio.sleep(10)  # 非阻塞
    return {"pong": True}

CPU 重活(转码、大计算)别指望线程池,GIL 帮不上忙。这类工作丢进程池或 Celery / RQ 一类队列更靠谱。

必须用同步 SDK 时,至少:

python
result = await asyncio.to_thread(sync_client.do_something, arg)

依赖注入不只是「注入数据库」

官方例子里依赖多半是拿 session、拿当前用户。生产里更值得把「校验」也做成依赖,避免每个接口复制一遍「帖子是否存在 / 是否本人」。

python
async def valid_post_id(post_id: UUID4) -> dict:
    post = await service.get_by_id(post_id)
    if not post:
        raise PostNotFound()
    return post

async def valid_owned_post(
    post: dict = Depends(valid_post_id),
    token_data: dict = Depends(parse_jwt_data),
) -> dict:
    if post["creator_id"] != token_data["user_id"]:
        raise UserNotOwner()
    return post

@router.put("/posts/{post_id}")
async def update_post(
    update_data: PostUpdate,
    post: dict = Depends(valid_owned_post),
):
    return await service.update(id=post["id"], data=update_data)

同一请求里依赖结果会缓存:parse_jwt_data 被链了三次,也只解码一次 JWT。

Pydantic 用狠一点,Settings 拆开一点

请求体别只写几个 str。正则、邮箱、枚举、长度上下限都能交给 Pydantic:

python
class UserBase(BaseModel):
    username: str = Field(min_length=1, max_length=128, pattern="^[A-Za-z0-9-_]+$")
    email: EmailStr
    age: int = Field(ge=18)

BaseSettings 建议按域拆:AuthConfig、全局 Config,别塞进一个巨型 Settings 类。环境变量一多,改密钥和改业务配置就会互相绊脚。

后台任务别滥用 BackgroundTasks

BackgroundTasks 适合发邮件回执、写日志这类短活,跟请求生命周期绑在一起。真正的异步作业(批量导出、视频处理、定时同步)还是上独立任务队列,失败重试、可观测性都更完整。

数据库这边:命名约定统一、Alembic 管迁移、测试客户端从第一天就用 async,能少踩很多后期重构。仓库也推荐用 ruff 统一格式和 lint。

怎么落地

  1. 打开 仓库 README(有 中文版
  2. 新项目直接按领域目录起骨架;老项目可以按模块逐步迁,不必一次搬完
  3. 如果团队在用编码代理,仓库还有一份机器可读的 AGENTS.md,方便塞进 Agent 上下文

这份清单是 opinionated 的,不是官方圣经。但「按领域拆 + 别堵事件循环 + 把校验做成依赖」这三条,几乎任何 FastAPI 生产项目都能立刻受益。

分享: