Python技术迷

告别臃肿!这个比Flash快10倍的Python框架,才是API开发的终极利器

那天晚上快十一点,我在公司楼下拿着奶茶刷接口压测报告,手机屏幕上那条红色的 95 分位延迟曲线,一路飙到 800 多毫秒,说实话有点窒息。 接口是 Flask 写的,一个很普通的用户查询接口,代码也不复杂,就是业务年头久了,中间垫了好几层,ORM、序列化、权限、日志一层一层叠上去,整个服务就开始显得有点“臃肿”。

后来我们把这条链路抽出来,用一个新的 Python 框架重写了一遍,压测数据直接把我干精神了: 同一台机器,同一个数据库,同样的数据量,QPS 提升了接近 10 倍,延迟从几百毫秒打到了几十毫秒以内。

就是今天要聊的这个主角:FastAPI。

先说说 Flask 用久了,哪儿容易“胖”起来

别误会啊,Flask 本身是个很优雅的框架,我自己也写了很多年。问题是——一旦项目活得久一点,它有几个典型现象:

  1. 路由散落在各个模块,蓝图嵌蓝图,想看一条完整调用链得点半天文件。
  2. 参数校验基本靠 if-else,或者自己封一层装饰器,时间久了没人敢动。
  3. 文档要么靠人手写,要么另外搞一套 Swagger 描述文件,接口一改就两边对不齐。
  4. 性能问题一来,你会发现同步阻塞链条很长:HTTP → Flask → 业务 → ORM → 外部服务,每一段都在等。

在流量不大、业务简单的时候,这些都不是事儿。但一旦你要做高并发 API 网关、BFF 层,或者跟前端约好“接口必须顶住每秒几千请求”,这些小问题就会不断放大。

所以我们当时就想:要不试试一个天然为异步、为高性能准备的框架?

这个“比 Flask 快一大截”的家伙到底是啥

简单讲一下背景,别太教科书:

  • 底座用的是 Starlette(一个高性能 ASGI 框架)
  • 数据校验用的是 Pydantic(基于类型标注,速度也不慢)
  • 跑在 uvicorn 这种异步服务器上(不是以前 Gunicorn+WSGI 那套)

这几个东西组合在一起,你就得到了 FastAPI。

它为什么会快?非常粗暴的理解就是: 以前 Flask 那种同步模型,处理一个请求要占一个线程(或者进程)一路等到数据库、RPC 都返回。 FastAPI 走异步模型,遇到 IO 等待就把控制权让出来,同一时刻可以有大量协程“挂起等待”,线程可以去干别的事。

同样一台 4 核 8G 的机器,Flask 开十几个 worker 跑得已经呼哧带喘,FastAPI 用 uvicorn 跑异步,往往还能从容多扛一截子流量。

先用一个最小的例子感受一下

我们先假装在写一个很简单的接口:健康检查 + 返回当前时间。代码大概这样:

# main.py
from datetime import datetime

from fastapi import FastAPI

app = FastAPI(title="demo-api")

@app.get("/ping")
asyncdefping():
return {"message": "pong"}

@app.get("/now")
asyncdefcurrent_time():
return {"now": datetime.utcnow().isoformat() + "Z"}

跑服务:

uvicorn main:app --reload --host 0.0.0.0 --port 8000

浏览器打开 http://localhost:8000/ping能看到:

{"message": "pong"}

更有意思的是,直接打开 http://localhost:8000/docs,会自动出现一套交互式文档,接口、参数、返回值一目了然,还能在线调试。这个在 Flask 里面,你要靠第三方扩展 + 自己维护一堆配置。

真正爽的在这:参数校验和数据模型

以前在 Flask 里写用户注册接口,参数校验大概是这样:

from flask import request, jsonify

@app.route("/users", methods=["POST"])
defcreate_user():
    data = request.get_json() or {}

if"name"notin data ornot data["name"]:
return jsonify({"error": "name is required"}), 400

if"age"notin data ornot isinstance(data["age"], int):
return jsonify({"error": "age must be int"}), 400

# 省略一堆 if-else

写多了人会崩溃,而且哪天字段多一个、类型改一下,逻辑就开始拧巴。

在 FastAPI 里,你直接把“数据长什么样”定义成一个 Pydantic 模型就完事了:

from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, Field

app = FastAPI()

classUserCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=32)
    age: int = Field(..., ge=0, le=150)
    email: Optional[EmailStr] = None
    is_admin: bool = False

classUserOut(BaseModel):
    id: int
    name: str
    email: Optional[str]

fake_db = {}
_id = 0

@app.post("/users", response_model=UserOut)
asyncdefcreate_user(user: UserCreate):
global _id
    _id += 1

    fake_db[_id] = user

return UserOut(
        id=_id,
        name=user.name,
        email=user.email,
    )

几个点随便说一下:

  • UserCreate 里面写的类型 + 校验规则,FastAPI 会自动用来校验请求体
  • 请求参数不符合要求,框架会直接返回 422,连错误信息都帮你组织好了
  • response_model=UserOut 告诉框架“返回值按这个模型来”,它会自动做过滤(比如不把密码这种字段带出去)

而且这一切会自动反映到接口文档里,前端一看 /docs 就知道这个接口该怎么调。

这类日常 CRUD 接口占了我们 API 开发工作量的大头,FastAPI 把这一块简化得非常狠。

来点更贴近实战的:拆业务、依赖注入、鉴权

真正的项目里不可能所有逻辑都堆在一个文件里,我们通常会拆成“路由 + service + 数据访问”几层。 FastAPI 这块比较顺手的一点,是它的“依赖注入”机制——你不用手动到处传数据库连接、当前用户之类的参数。

随便举一个比较贴近日常的例子: 我们做一个需要校验 Token 的用户信息接口。

# auth.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

asyncdefget_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
)
:

    token = credentials.credentials
# 这里简化处理,真实情况你会解析 JWT、查缓存之类
if token != "token-123":
raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="invalid token",
        )
# 假装这是一个用户对象
return {"id": 1, "name": "dongge"}

# main.py
from fastapi import Depends, FastAPI

from auth import get_current_user

app = FastAPI()

@app.get("/me")
asyncdefme(current_user=Depends(get_current_user)):
return current_user

这段逻辑有几个隐性的好处:

  1. get_current_user 可以在很多接口重复用,鉴权逻辑集中在一处
  2. 不需要在每个接口手动从 header 里拿 token
  3. 单元测试的时候可以非常方便地把 get_current_user 换成一个假的实现(依赖覆盖)

以前在 Flask 里,大家一般是自己写装饰器来做这些事,但装饰器一旦多起来,调试和组合就没那么直观了。

“比 Flask 快 10 倍”这句话,落到指标上大概是啥样

这个事我一开始也挺怀疑的,直到真上了压测。

我们当时选了一条比较典型的接口:

  • 读写一次 Redis
  • 查一条 MySQL
  • 做一点轻度 JSON 处理

硬件配置:4C8G,跑在容器里。 两种实现:

  • 版本 A:Flask + gunicorn(sync worker)
  • 版本 B:FastAPI + uvicorn(async),数据库用异步驱动

非常粗糙地贴一下结果(单位就不写那么严谨了,你感受下就行):

  • 单机 QPS:

    • Flask 版本:稳定在 800 左右就开始抖
    • FastAPI 版本:往上推到接近 8000 才开始出现 95 分位延迟明显抬头
  • P95 延迟:

    • Flask:七八百毫秒
    • FastAPI:稳定在一百多毫秒以内

这还不是极限调优,只是“正常人能接受的优化力度”,外面那种实验室级别的压测通常会把数字堆得更夸张。 所以所谓“快 10 倍”,更多是一个量级概念——只要你的接口主要瓶颈在 IO 上(比如查库、调其他服务),FastAPI 的异步模型就有发挥空间。

当然,如果你的业务全是 CPU 密集型计算(大量图像处理、PDF 渲染之类),那就别指望框架本身能帮你多少,该上多进程还是得上。

老项目要迁移,大概该怎么下手

真要从 Flask 迁到 FastAPI,我的建议是不搞那种“一刀切式的 Big Bang 重构”,而是:

  1. 先新接口用 FastAPI 写,比如新起一个 v2 版本
  2. 在网关层或者 Nginx 里分流:老接口还是到 Flask,新接口走 FastAPI
  3. 共用的业务逻辑(比如 service 层、DAO 层)尽量抽到独立模块,让两个框架都能 import
  4. 等新接口稳定了,把流量慢慢切回来,把老的逐步下线

举个非常简化的结构例子:

# biz/service/user_service.py
classUserService:
def__init__(self, repo):
        self.repo = repo

asyncdefget_user(self, user_id: int):
# 这里用的是异步仓储(比如 asyncpg)
returnawait self.repo.get(user_id)

# api_fast/user_api.py
from fastapi import APIRouter, Depends

from biz.service.user_service import UserService
from infra.repo import get_user_repo  # 一个返回 repo 实例的依赖函数

router = APIRouter(prefix="/users")

@router.get("/{user_id}")
asyncdefget_user(
    user_id: int,
    service: UserService = Depends(
        lambda repo=Depends(get_user_repo): UserService(repo)
    )
,
)
:

    user = await service.get_user(user_id)
return user

Flask 那边也可以 import UserService,只不过调用的时候用 asyncio.run() 或者干脆给它写一个同步版本的 repo。 这样迁移的过程比较平滑,心理压力也小很多。

性能好,坑也有,提前说清楚

任何框架都不是银弹,FastAPI 也一样,有几个坑我见人踩得比较多:

  1. 写着写着,就开始在 async def 里调用一堆同步阻塞库(同步 MySQL、同步 HTTP 客户端),结果整个事件循环卡死,性能瞬间打回 Flask 时代。 要么用异步驱动库,要么把重 IO 的逻辑丢到线程池里:

    import asyncio

    import httpx
    from fastapi import FastAPI

    app = FastAPI()
    client = httpx.Client()  # 同步客户端

    defblocking_call():
        resp = client.get("https://example.com")
    return resp.text

    @app.get("/external")
    asyncdefcall_external():
        data = await asyncio.to_thread(blocking_call)
    return {"len": len(data)}

  2. 滥用全局变量,尤其是数据库连接、客户端实例,这个在异步环境下更容易整出一些奇怪的线程安全问题。比较推荐的写法是:在 startup 事件里初始化,在 shutdown 里统一关闭。

    from fastapi import FastAPI

    app = FastAPI()

    db = None

    @app.on_event("startup")
    asyncdefstartup():
    global db
    # 伪代码:初始化异步数据库连接池
        db = await create_db_pool(dsn="...")

    @app.on_event("shutdown")
    asyncdefshutdown():
    await db.close()

  3. 类型标注乱写、随便用 dict 来回传。 这样短期看没啥问题,但你会失去一大半 FastAPI 的优势(自动文档、自动校验、IDE 提示),等项目跑大了再补类型,基本没人有勇气动手。

写到这差不多,你大概心里有个谱了

如果你现在手上有个 API 服务:

  • 请求量不小,
  • 经常因为某几个慢接口把整个服务拖得很难看,
  • 而且刚好是 Python 写的,

那真可以抽个晚上,拿其中一条链路用 FastAPI 改一版压一下测。 不需要一上来就 all in,把最核心、最痛的那几个接口先挑出来,效果往往比你想象中直观。