FastAPI 用下来的一些感受

之前团队一直用 Flask，新起的几个内部服务这次换成了 FastAPI，图的是自带的 pydantic 校验和自动生成的 Swagger 文档能省不少事。用了几个月，整体是满意的，但也不是网上吹的那么零成本，记几个印象比较深的点。

最直观的好处就是接口文档不用自己维护了，路由函数写好类型注解，/docs 那个 Swagger UI 就自动生成出来了，前端同事可以直接对着这个页面联调，比以前手写 API 文档、经常忘记同步的日子舒服太多。

from pydantic import BaseModel

class OrderCreate(BaseModel):
    user_id: int
    items: list[str]
    remark: str | None = None

@app.post("/orders")
async def create_order(order: OrderCreate):
    return await order_service.create(order)

请求体校验也是这套 pydantic model 顺带做了，类型不对、字段缺失，框架自动 422 返回，不用手写一堆 if not isinstance 的校验代码，这块确实爽。

不过用起来最容易踩的坑还是 async/sync 混用这块。团队里有几个同事之前写 Flask 惯了，习惯性地把数据库查询、文件读写这种阻塞操作直接放进 async def 的路由里，写出来大概是这样：

@app.get("/report")
async def get_report():
    data = sync_db_client.query("SELECT * FROM big_table")  # 同步阻塞调用
    return process(data)

这段代码能跑，但跑起来会把整个 event loop 卡住，因为 sync_db_client.query 是阻塞的同步调用，在 async def 里直接调用不会被丢到线程池，而是直接占用事件循环的这一条线程，高并发的时候这一个接口能把其他所有协程都堵住，QPS 直接腰斩。压测的时候纳闷怎么 async 框架比之前 Flask 还慢，查了半天才反应过来是这个问题。

解法要么把这类阻塞调用扔进线程池：

from starlette.concurrency import run_in_threadpool

@app.get("/report")
async def get_report():
    data = await run_in_threadpool(sync_db_client.query, "SELECT * FROM big_table")
    return process(data)

要么干脆路由函数就别写成 async def，普通的 def 路由 FastAPI 会自动放到线程池里跑，对纯同步代码反而更合适，不用自己包一层。团队后来定了个规矩：路由里但凡调用的是同步的库（大部分老的 ORM、老的 SDK 都是同步的），路由函数就老老实实写 def 不写 async def，混用的坑才算彻底避开。

依赖注入这块（就是 Depends）一开始觉得挺优雅，权限校验、分页参数解析都能抽成公共依赖，写多了之后又觉得有点用力过猛，一个复杂点的接口能叠七八层 Depends，调试的时候跳来跳去找不到某个值到底哪来的，尤其是嵌套依赖，A 依赖 B、B 又依赖 C，出问题得一层层往回倒。这个不算 FastAPI 的锅，更多是团队自己写的时候没控制住抽象层数，但确实是个容易失控的地方，用的时候得克制一下，不是什么逻辑都要塞进 Depends。

另外后台任务 BackgroundTasks 这个东西刚开始以为能顶一部分异步任务队列的活，用了才发现它就是请求返回之后在同一个进程里接着跑一段代码，进程重启这个任务就没了，完全没有持久化和重试，稍微正经点的异步任务该用 Celery 还是老老实实用 Celery，BackgroundTasks 撑死了适合发个通知邮件这种失败了也无所谓的场景。

整体来说从 Flask 换过来，效率提升是真实的，尤其是文档自动生成和请求校验这两块，团队协作成本降了不少。就是 async 这套东西对之前只写惯同步代码的人不算无脑好上手，容易在不知不觉中写出阻塞 event loop 的代码，得过一遍这个坑才能真正上手顺畅。

FastAPI 用下来的一些感受

Post Actions

Navigation

More Posts