给三年前的老 Flask 项目换心脏：一次 FastAPI 迁移实录

手头有个跑了三年的 Flask 项目，接口一百多个，代码两万来行。最近需求变多，同步阻塞的架构开始吃不消——导出报表的接口一跑，整个服务跟着卡。趁这次迭代，我把它迁到了 FastAPI，把过程和踩的坑记录下来。

为什么要迁

三个理由，按权重排序：

1. 异步能力。项目里有大半接口在等外部服务：查数据库、调第三方 API、生成文件。Flask 同步模型下每个请求占一个 worker，等外部响应的时间全浪费了。FastAPI 基于 asyncio，等待时间可以让出去干别的。
2. 文档自动化。Flask 里接口文档靠手写 Markdown，改接口经常忘了同步文档。FastAPI 自带 OpenAPI 生成，接口签名即文档。
3. 类型提示。Pydantic 的请求校验比手写 if-else 干净太多，一个声明抵十几行防御代码。

第一步：别急着换框架，先把路由层解耦

老代码的典型问题：视图函数直接操作 request 和数据库连接。直接换框架等于重写。

我的做法是先加一层薄抽象。把每个视图函数拆成"解析参数"和"业务函数"两段：

# 迁移前：Flask 风格
@app.route("/api/orders", methods=["GET"])
def list_orders():
    status = request.args.get("status")
    page = int(request.args.get("page", 1))
    rows = db.query("SELECT * FROM orders WHERE status=%s LIMIT %s OFFSET %s",
                    status, PAGE_SIZE, (page - 1) * PAGE_SIZE)
    return jsonify(rows)

# 中间态：业务函数与框架解耦
def list_orders_service(status: str, page: int) -> list[dict]:
    return db.query("SELECT ...", status, PAGE_SIZE, (page - 1) * PAGE_SIZE)

# Flask 壳
@app.route("/api/orders")
def list_orders():
    return jsonify(list_orders_service(request.args.get("status"), int(request.args.get("page", 1))))

这一步是纯重构，测试全绿后才动 FastAPI。中间态大概花了一周，但值得——后面换壳时业务函数一行没改。

第二步：FastAPI 的壳

from fastapi import FastAPI, Depends, Query
from pydantic import BaseModel, Field

app = FastAPI(title="订单服务")

class OrderOut(BaseModel):
    id: int
    status: str
    amount: float = Field(..., description="订单金额")

@app.get("/api/orders", response_model=list[OrderOut])
async def list_orders(
    status: str = Query("all", description="订单状态"),
    page: int = Query(1, ge=1),
):
    rows = await run_in_pool(list_orders_service, status, page)
    return rows

同步的 list_orders_service 没有重写，用 run_in_pool（包了一层 asyncio.to_thread）丢进线程池。这是迁移期的过渡手段，收益立竿见影：报表导出再也不会堵死其他接口。

踩的三个坑

坑一：全局状态在异步下不安全。 老代码用模块级字典做简易缓存，Flask 多进程下各写各的没出过事；FastAPI 单进程多协程后，出现了两个请求同时改同一个 key 导致的数据错乱。后来换成了 cachetools.LRUCache 加锁，再后来上了 Redis。

坑二：SQLAlchemy 版本陷阱。 项目用的是 SQLAlchemy 1.4 的老式 Query API。在异步场景下想让异步引擎接管，得升到 2.0 风格并用 AsyncSession。我一开始图省事继续用同步 Session 加线程池，结果连接池被线程打爆。正确的姿势：要么彻底切异步引擎，要么维持纯同步部署，半吊子混搭最危险。

坑三：依赖注入的顺序。 FastAPI 的 Depends 是嵌套解析的，我写了一个依赖链里某个函数去读请求体，导致另一个同时消费请求体的接口直接读不到——请求体流只能消费一次。这种问题文档里提了一嘴，但不踩一次真不知道。

收尾

迁移完成后的压测数据：同样 4 核 8G 的容器，等待外部服务的场景下，QPS 从 120 提到 470，P99 从 2.3 秒降到 680 毫秒。数字不能全归功于框架——中间态重构时顺手把几条烂 SQL 修了——但异步模型确实把"等"的时间利用起来了。

如果有人也在维护一个年久失修的 Flask 项目，我的建议是：先做路由层解耦，测试覆盖到位之后再动框架。直接重写是最容易失败的路，渐进迁移虽然慢一点，但每一步都能上线。