FastAPI 这几年在 Python 后端圈子里基本已经是默认选项了。我最早是在 2019 年底一个内部工具项目里用它,那时候 Flask 还是主流,FastAPI 刚发布没多久,文档也不像现在这么全。后来陆陆续续用它做了十几个 API 服务,从企业内部的小工具到日请求量百万级的线上服务都有,算是把它的脾性摸得比较透。这篇文章不聊官方文档里有的东西,主要讲讲我实际项目里怎么组织目录、怎么压性能、怎么部署、怎么把大模型接进来,以及那些文档不会告诉你的坑。
适合谁看?想从 Flask/Django 转过来的后端开发,准备用 FastAPI 做正式项目的同学,以及已经在用但总觉得哪里没搞对的人。我会尽量把每一步都讲清楚为什么这么做,而不是甩一段代码让你自己去猜。
1. 先搞清楚:FastAPI 凭什么成为现代 API 的首选
1.1 它解决的三个核心痛点
先说第一个痛点:异步支持。Python 的 GIL 让很多人对并发有心理阴影,但 FastAPI 从底层就是为 asyncio 设计的。它的请求处理可以完全跑在事件循环上,一个 worker 能扛住的并发连接数远超传统的多线程模型。我用同一个服务做过对比,同步写法下压测 QPS 大概 800 左右,改成 async 之后直接冲到 2500+,这还是没做任何其他优化的情况下。
第二个痛点是数据校验。以前写 Flask 接口,参数校验全靠手写 if 判断,写多了自己都烦。FastAPI 把 Pydantic 直接嵌进框架里,你在类型注解里写清楚参数类型和结构,框架自动帮你完成解析、校验、转换。类型错了返回 422 而不是 500,参数缺了直接报具体错误位置,这些在传统框架里都要自己造轮子。
第三个痛点是文档。FastAPI 基于 OpenAPI 标准,只要你把路由和 Pydantic 模型写出来,Swagger UI 和 ReDoc 就直接生成了。这个好处在团队协作和前后端联调时特别明显,前端同事不用追着你问字段含义,打开 /docs 自己看就行。
1.2 和 Flask、Django 怎么选
我整理了一张选型参考表,基于我实际使用的感受:
| 框架 | 异步支持 | 数据校验 | 文档生成 | 生态成熟度 | 适合场景 |
|---|---|---|---|---|---|
| FastAPI | 原生 async/await | Pydantic 自动化 | 自动 OpenAPI | 快速增长,但不如老牌 | 高并发 API、微服务、AI 服务 |
| Flask | 需要插件(quart 等) | 手写校验 | 手动配置 | 非常成熟 | 简单项目、老系统维护 |
| Django + DRF | 支持但较重 | Serializer 半自动 | 半自动 | 非常成熟 | 含后台管理、ORM 全家桶的复杂 Web 系统 |
我的习惯是:纯 API 服务、尤其是要对接大模型或者做实时数据推送的,无脑选 FastAPI。如果项目里有复杂的后台管理界面、需要现成的 Admin 系统,那 Django 可能更省事。但如果是新项目且前后端分离,FastAPI 完全够用,而且后面的维护成本是真的低。
2. 项目目录结构:决定这套 API 能活多久
2.1 我推荐的分层结构
很多人用 FastAPI 还是 Flask 的思维,一个 main.py 里写几百行路由。这在 demo 里没问题,但项目过两周你就想重写。我踩过这个坑,后来整理出一套结构,已经在多个生产项目里验证过,可以照抄:
fastapi_project/ ├── app/ │ ├── __init__.py │ ├── main.py # 应用入口,创建 FastAPI 实例 │ ├── core/ │ │ ├── config.py # 配置管理(pydantic-settings) │ │ ├── security.py # 鉴权逻辑(JWT、API Key) │ │ └── logging.py # 日志配置 │ ├── api/ │ │ ├── __init__.py │ │ ├── v1/ │ │ │ ├── __init__.py │ │ │ ├── router.py # v1 版本路由聚合 │ │ │ ├── endpoints/ │ │ │ │ ├── users.py │ │ │ │ └── orders.py │ │ │ └── schemas/ # Pydantic 请求/响应模型 │ │ └── deps.py # 公共依赖(当前用户、DB Session) │ ├── models/ # SQLAlchemy ORM 模型 │ ├── services/ # 业务逻辑层 │ ├── repositories/ # 数据访问层(可选) │ └── utils/ # 工具函数 ├── tests/ # pytest 测试 ├── alembic/ # 数据库迁移(若用 alembic) ├── pyproject.toml ├── .env # 本地环境变量(不提交到 Git) └── Dockerfile核心原则是按业务域分层,而不是按文件类型分层。你要把 users 相关的路由、schema、业务逻辑放在一起,而不是把所有路由塞一个文件、所有模型塞另一个文件。改一个功能时只需要在同一个目录里动,排查问题也只要顺着目录名找。
2.2 配置管理:别把密钥写在代码里
配置这块我吃过亏。早期项目把数据库连接串直接写在 config.py 里,结果代码传到私有仓库,后来改了密码全项目都要改。现在用 pydantic-settings 管理配置,强类型、自动从环境变量读取,非常省心:
from pydantic_settings import BaseSettings class Settings(BaseSettings): app_name: str = "MyAPI" database_url: str = "postgresql+asyncpg://user:pass@localhost/db" redis_url: str = "redis://localhost:6379/0" jwt_secret_key: str = "change-me-in-prod" jwt_expire_minutes: int = 60 * 24 # 大模型相关配置(后面会用到) llm_base_url: str = "https://api.deepseek.com/v1" llm_api_key: str = "" ollama_base_url: str = "http://localhost:11434" model_config = {"env_file": ".env", "env_file_encoding": "utf-8"} settings = Settings()在 FastAPI 的依赖里注入这个 settings 单例,比到处from config import settings好测试、好替换。注意.env文件一定不要提交到 Git,里面全是敏感信息,提交一次后面就得改密钥。
3. 核心特性实操:路由、Pydantic 与依赖注入
3.1 路由与参数处理的正确姿势
FastAPI 的路由声明很直观,但很多人忽略了一个点:参数顺序和默认值会影响语义。看这个例子:
from fastapi import FastAPI, Query, Path, Body app = FastAPI() @app.get("/api/v1/users/{user_id}/orders") async def get_user_orders( user_id: int = Path(..., description="用户ID"), status: str | None = Query(None, pattern="^(pending|paid|shipped)$"), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), ): return {"user_id": user_id, "status": status, "page": page, "page_size": page_size}这里有几个细节。Path(..., ...)的第一个参数是...,表示必填,这个必须记住,漏了的话 user_id 会变成可选参数,类型标错都不报错。Query(pattern=...)可以内联正则校验,比在函数里写 if 干净得多。ge/le是数值范围校验,FastAPI 会自动帮你拦截非法值。
说到路由组织,一定要学会APIRouter。不同的业务模块建各自的 router,然后在主路由里include_router,服务一多你就知道这个多重要了。
3.2 Pydantic 模型:请求与响应分离
Pydantic 模型是 FastAPI 的精华。我最想强调的一点是:请求模型和响应模型一定要分开写。很多新手图省事,同一个模型既用来接收请求又用来返回响应,结果就是密码字段返回给前端、内部字段暴露出去。正确做法是三个模型:
from pydantic import BaseModel, EmailStr, Field class UserCreateRequest(BaseModel): username: str = Field(min_length=3, max_length=50) email: EmailStr password: str = Field(min_length=8) class UserResponse(BaseModel): id: int username: str email: EmailStr created_at: datetime class UserUpdateRequest(BaseModel): username: str | None = None email: EmailStr | None = None响应模型在路由里用response_model声明,FastAPI 会自动做数据过滤和序列化。比如你的 ORM 对象里有password_hash字段,只要它不在UserResponse里,返回时就会被丢掉,不需要你手动删。
另外一个技巧是model_config = {"from_attributes": True}。把 ORM 对象直接传给 Pydantic 模型做响应转换时,有这个配置才能正常读取对象属性,否则只认 dict。我每次新建模型都会写上一句,省得后面到处踩坑。
3.3 依赖注入:把鉴权和数据库会话交给框架
依赖注入是 FastAPI 最容易被人忽视的能力。它的语法其实就是函数参数里声明依赖,框架自动帮你解析。用途很多:鉴权、数据库会话、配置注入、请求上下文。
举一个数据库会话的例子。用 SQLAlchemy 2.0 的 async 版本时,我会在app/api/deps.py里定义:
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine engine = create_async_engine(settings.database_url, echo=False, pool_size=10, max_overflow=20) AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False) async def get_db() -> AsyncIterator[AsyncSession]: async with AsyncSessionLocal() as session: yield session然后在路由里:
from fastapi import Depends @app.get("/api/v1/users/me") async def get_me( current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): ...get_current_user是另一个依赖,它从请求头里解析 JWT、查库、返回当前用户。FastAPI 会自动判断依赖之间的嵌套关系,get_me需要get_current_user和get_db,框架就会先执行这两个依赖。这个机制让鉴权和资源管理变得极其干净——每个依赖只干一件事,组合起来就是完整的请求流程。
注意一个性能细节:如果依赖函数是普通def(不是async def),FastAPI 会把它丢到线程池里执行;如果是async def,才会跑在事件循环上。所以耗时的同步操作(比如查询同步数据库)用普通def反而更好,不会阻塞事件循环。
4. 性能优化:把 FastAPI 的异步优势真正用起来
4.1 async def 和 def 的选择,决定了你的 QPS
这是 FastAPI 最核心的调优点之一,也是面试最爱问的。简单说:
- 路由处理函数里如果有阻塞型 IO(同步数据库驱动、requests 库调用外部 HTTP、文件读写),用普通
def声明,让 FastAPI 把它放到线程池。 - 如果全链路都是非阻塞异步(asyncpg、httpx.AsyncClient、aiofiles),用
async def。 - 最忌讳的是:
async def函数里调requests.get或time.sleep,这会把整个事件循环卡死,所有并发请求全部排队。
我举个反面教材。早期项目里有段代码:
@app.get("/api/v1/health") async def health(): time.sleep(2) # 模拟一个同步耗时操作 return {"status": "ok"}实测并发 10 个请求,后面 9 个全部要等第一个睡完才进来,总耗时 20 秒。改用def声明后,FastAPI 会把每个请求丢到线程池,耗时变成了 2 秒。就这么一个关键字的变化,效果天差地别。
4.2 数据库连接池与并发控制
数据库连接池是性能瓶颈的重灾区。很多人连 SQLAlchemy 直接不配连接数,默认 5 个连接,稍微来点并发就报TimeoutError。我在生产环境的经验值是:
engine = create_async_engine( settings.database_url, pool_size=20, max_overflow=30, pool_timeout=30, pool_pre_ping=True, )pool_pre_ping=True特别重要。数据库重启后,旧连接全部失效,没有 pre_ping 的话第一条查询必报Connection is closed。加了它,每次拿连接前先验证一下,省掉一堆诡异报错。
还有一点:不要把事务开得太久。FastAPI 的请求生命周期里,尽早查库、尽早 commit、尽早释放连接。我在get_db依赖里用async with,就是为了保证请求结束一定能关掉连接。连接池一旦被占满又没人释放,整个 API 就变成"看起来还活着但什么都查不了"的状态。
4.3 加一层 Redis 缓存,接口延迟直接减半
对于读多写少的接口,Redis 缓存是见效最快的优化手段。我一般在 service 层做缓存,而不是在路由层,这样逻辑更清晰:
import json import httpx from fastapi.encoders import jsonable_encoder async def get_user_profile_with_cache(user_id: int) -> dict: cache_key = f"user:profile:{user_id}" cached = await redis_client.get(cache_key) if cached: return json.loads(cached) profile = await fetch_user_profile_from_db(user_id) await redis_client.set(cache_key, json.dumps(jsonable_encoder(profile)), ex=300) return profileTTL 设 300 秒,热点数据 5 分钟过期一次,既不会太陈旧,也不会占用太多内存。注意缓存穿透问题——如果查询的 user_id 不存在,一定要把空结果也缓存几秒钟,否则恶意请求可以直接把你数据库打到趴下。
4.4 压测工具与实际调优经验
压测不能只报一个 QPS 数字。我用 wrk 和 Locust 都有,简单说下流程。wrk 适合快速摸底:
wrk -t8 -c200 -d30s http://localhost:8000/api/v1/users/me我调优时关注的指标依次是:P90 延迟 < 100ms、错误率 < 0.1%、CPU 不要被打满。如果延迟高但 CPU 不高,大概率是 IO 阻塞(检查有没有同步调用);如果 CPU 满但 QPS 上不去,大概率是 GIL 竞争或者序列化太重(检查 Pydantic 模型字段是不是太多、日志打印是不是太频繁)。
压测还有个容易忽略的点:本机压本机,数据会好看 30%。上线前要用独立压测机打,网络延迟才真实。
5. 接上大模型:FastAPI 的现代 API 玩法
5.1 让 FastAPI 调用本地 Ollama 模型
现在很多人把 FastAPI 当大模型应用的底座。我自己做过一个项目,FastAPI 作为统一 API 网关,背后接 Ollama 本地模型和云端大模型 API,前端只需要跟一个地址通信。
接 Ollama 很简单,如果只是同步等待结果:
import httpx async def ask_ollama(prompt: str) -> str: async with httpx.AsyncClient(timeout=120) as client: resp = await client.post( f"{settings.ollama_base_url}/api/generate", json={ "model": "qwen2.5:7b", "prompt": prompt, "stream": False, }, ) resp.raise_for_status() return resp.json().get("response", "")注意 timeout 一定要调大。本地模型推理速度取决于显卡,7B 模型生成几百个 token 可能要十几秒,默认的 5 秒超时一定会断。
5.2 流式响应:让 Tokens 一个字一个字蹦出来
大模型 API 的流式响应是刚需。用户不喜欢等十几秒才看到第一个字。FastAPI 可以用StreamingResponse实现 SSE(Server-Sent Events):
from fastapi.responses import StreamingResponse import json @app.post("/api/v1/chat/stream") async def chat_stream(payload: ChatRequest): async def event_generator(): async with httpx.AsyncClient(timeout=None) as client: async with client.stream( "POST", f"{settings.llm_base_url}/chat/completions", json={ "model": payload.model, "messages": payload.messages, "stream": True, }, headers={"Authorization": f"Bearer {settings.llm_api_key}"}, ) as resp: resp.raise_for_status() async for line in resp.aiter_lines(): if line.startswith("data:"): data = line.removeprefix("data:").strip() if data == "[DONE]": yield "data: [DONE]\n\n" break try: chunk = json.loads(data) delta = chunk["choices"][0]["delta"].get("content", "") if delta: yield f"data: {json.dumps({'delta': delta})}\n\n" except json.JSONDecodeError: continue return StreamingResponse(event_generator(), media_type="text/event-stream")这里有两个细节。第一,httpx.AsyncClient(timeout=None)必须配,流式读取的时长是由输入端决定的,不能设固定超时。第二,SSE 格式要求每帧以data:开头、以空行结尾,格式错了前端 EventSource 解析不出来。
5.3 基于 FastAPI + LangChain/LangGraph 的 Agent 服务
如果要把 AI Agent 真正落地到生产,不只是做一个聊天接口,我建议把 Agent 编排和 API 层分离。我在项目中把 LangGraph 的 Agent 逻辑封装成一个独立的 service,FastAPI 只负责接收请求、鉴权、限流、把参数传给 Agent、再把结果流式推回前端。
这样做的优势很明显:Agent 的循环、工具调用、状态管理在 LangGraph 里维护,API 层的职责纯粹是承载流量。哪一天你觉得 LangGraph 太重想自己写编排,只需要改 service 层,路由和模型都不用动。之前我做过一个给客服团队用的工单分析 Agent,就是 FastAPI 接 LangGraph,后台异步跑工具链,跑完推送结果到客户端,整个链路非常顺。
顺便说一句,FastAPI 不是 Gradio 的替代品。Gradio 适合快速演示模型效果,FastAPI 适合做生产级服务。如果你要接微信公众号这类第三方平台,需要稳定的接口和自定义鉴权,那一定是 FastAPI 的活。Gradio 做原型,FastAPI 做线上,两不冲突。
6. 部署与打包:从开发机到生产环境
6.1 Uvicorn + Gunicorn 的正确姿势
开发时uvicorn app.main:app --reload就够了,但生产环境不能这么跑。uvicorn 支持--workers多进程,不过更标准的是用 Gunicorn 做进程管理器,让 uvicorn 做 worker:
gunicorn app.main:app \ -k uvicorn.workers.UvicornWorker \ -w 4 \ -b 0.0.0.0:8000 \ --timeout 120 \ --access-logfile - \ --error-logfile -关键点:-k必须指定uvicorn.workers.UvicornWorker,否则 Gunicorn 会用默认的同步 worker,你的异步代码直接废掉。-w 4是 worker 数,经验公式是CPU 核心数 × 2 + 1,不是越多越好。worker 数太多会导致上下文切换开销大于收益,我用 8 核机器跑 4 个 worker,性能反而比 8 个稳定。
6.2 用 Docker 打包部署
Dockerfile 我贴一个生产可用的:
FROM python:3.11-slim AS base WORKDIR /app ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD ["gunicorn", "app.main:app", "-k", "uvicorn.workers.UvicornWorker", "-w", "4", "-b", "0.0.0.0:8000"]注意PYTHONUNBUFFERED=1必须设。Python 的 print 默认是块缓冲的,在 Docker 容器里日志会滞后甚至丢失,设了这个环境变量日志才能实时输出。这是很多人容器里看不到日志的直接原因。
镜像不要塞进不需要的文件。我通常会在项目根目录加一个.dockerignore,把.git、tests、__pycache__、.venv全排除掉,镜像体积能小一半,构建速度也快很多。
6.3 Windows 打包成可执行文件
有人问 FastAPI 项目能不能在 Windows 上打包成单个 exe。可以,但我强烈建议你想清楚场景。如果你是给内部工具、给不会装 Python 的同事用,那 PyInstaller 可行;如果是要部署到服务器,别这么干,直接 Docker。
如果一定要打包,有几个必踩的坑先说在前面。PyInstaller 默认不会收集 uvicorn 和你的应用模块,你需要手动指定:
pyinstaller --name myapi --collect-all uvicorn --collect-all app main.pymain.py里要把app实例化逻辑放在入口,避免循环导入。还有,打包后要记得附一个config.yaml或让程序读取旁边的.env文件,因为路径变了,默认配置可能找不到。
6.4 uvicorn 日志丢失问题排查
热词里那个 "uvicorn fastapi 日志丢失问题" 我太熟了,至少遇到三次。症状是:接口偶尔报错,但看日志什么都没有;或者 Gunicorn 起来之后,Python 代码里的 print 全都不见了。
原因有三个。第一个是前面说的缓冲问题,Python 的 stdout 缓冲,尤其是被 Gunicorn 接管后,print 的内容攒在缓冲区里没刷出来。解决方案是启动参数加--capture-output或者设置PYTHONUNBUFFERED=1。
第二个是 uvicorn 的 access log 和自定义日志混在一起,格式乱、看不出顺序。我建议在生产中单独配 logging 文件,把 FastAPI 的uvicorn.accesslogger 和业务 logger 分开处理:
import logging # 在 app/core/logging.py 中 def setup_logging(): logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s - %(message)s", handlers=[ logging.FileHandler("app.log", encoding="utf-8"), logging.StreamHandler(), ], ) logging.getLogger("uvicorn.access").setLevel(logging.WARNING)第三个是异常被吞了。FastAPI 里有些异常在 StreamingResponse 生成器内部发生,响应已经开始就没法返回错误状态码了,异常只打印在 stderr。如果 stderr 没被收集,日志就"丢"了。这个需要把 stderr 也重定向到日志文件,或者用 Sentry 之类的错误收集服务兜底。
7. 踩坑实录与面试考点速查
7.1 常见故障与解决办法
我把实际项目中最常遇到的问题整理成一张速查表:
| 问题 | 典型原因 | 解决方案 |
|---|---|---|
| 接口大量 500,数据库连接报错 | 连接池耗尽或连接失效 | 调整pool_size,加pool_pre_ping=True |
| 并发一上来就卡死 | async 函数里用了同步阻塞调用 | 改成def或换异步客户端 |
| 日志没有输出 | stdout 缓冲 / stderr 未捕获 | 设PYTHONUNBUFFERED=1,重定向 stderr |
| CORS 跨域报错 | 前端域名不在允许列表 | 用CORSMiddleware配置allow_origins |
| 上传大文件超时 | 代理层超时设置太短 | 调整 Nginx/Gunicorn 的 timeout 参数 |
| Pydantic 校验不过但报 500 | from_attributes没配 | 在响应模型加model_config = {"from_attributes": True} |
| Docker 里连不上宿主机数据库 | localhost 指向容器自身 | 用host.docker.internal或配置真实 IP |
还有一个坑是time.sleep在 async 代码里出现。团队里有人从 Flask 转过来,习惯性在async def里写time.sleep(1),整个服务的所有请求瞬间被卡成串行。查找办法很简单:压测时看到延迟呈线性累积,基本就是这个原因。
7.2 FastAPI 面试题汇总
面经相关的热词也很多,我结合当面试官的经验,经常考这几个问题:
Q:FastAPI 为什么快?答:基于 Starlette 的异步能力 + Pydantic 的高性能解析 + 自动化的数据校验。它把每个请求的处理尽量放到事件循环上,而不是比 Flask 多做了多少黑魔法。
Q:async def和def的区别?答:def路由会被放到线程池执行,适合阻塞型 IO;async def在事件循环执行,适合非阻塞异步。混用时要特别注意不要阻塞事件循环。
Q:依赖注入如何实现?答:FastAPI 通过分析函数的签名和参数默认值,自动解析依赖树。每个依赖可以是函数或类,支持嵌套、缓存(@lru_cache)和请求作用域。
Q:如何管理数据库会话?答:用依赖yield一个 Session,请求结束时自动关闭。推荐 SQLAlchemy 2.0 async 版本,配合连接池配置。
Q:如何做限流?答:FastAPI 没内置限流,可以用中间件配合 Redis 计数实现,或者用 slowapi 这类第三方库。生产环境更推荐在网关层(如 Nginx、云厂商 API 网关)做。
7.3 我的几个独家小建议
最后分享几个不太容易在文档里看到的心得。
第一,路由前缀整体规划。我在app/main.py里会统一配置:
app = FastAPI(title="MyAPI", version="1.0.0", docs_url="/docs", redoc_url=None) app.include_router(api_router, prefix="/api/v1")redoc_url=None可以关掉不用的文档,少暴露一个端点。docs_url在正式环境也可以考虑关掉或加鉴权,避免接口结构被外部看到。
第二,Pydantic 模型字段别设计太满。响应字段越多,序列化越慢。我之前有个接口为了通用性返回了 40 多个字段,后来拆成精简版和完整版两个模型,流量大的场景用精简版,整体响应时间直接降了一半。
第三,写测试的优先级比你想的高。FastAPI 的TestClient基于 httpx,测试起来很顺手。我给自己定的规则是:每个新增接口至少配一个成功用例、一个校验失败用例、一个鉴权失败用例。有了这层保障,后面做性能调优时才敢大胆改代码。
写在最后
用 FastAPI 这几年,我的感受是:它是一个把"正确的事情"变成"默认的事情"的框架。类型注解写了,校验和文档就都有了;异步写对了,性能就上来了;依赖注入用好了,代码就干净了。大部分性能问题其实不是 FastAPI 的问题,而是使用方式的问题。
如果这篇文章对你有一点用,建议你拿一个小项目亲手练一遍:用我上面的目录结构搭一个服务,接上数据库和 Redis,加一个流式接口,再压一次测。走完这个流程,你对 FastAPI 的理解会超过大多数只在文档里看过它的人。遇到什么没写到的坑,欢迎在评论区一起交流。