FastAPI 是当下最火的 Python 异步 Web 框架，凭借类型提示驱动的数据校验、原生 async/await 并发和自动生成的 OpenAPI 文档，成为构建高性能 API 服务的首选。它不像 Flask 那样需要自己拼装扩展，也不像 Django 那样背负整套全家桶，而是把”写接口”这件事收敛成纯粹的 Python 函数加类型注解。本文从零带你落地一个可上生产的 FastAPI 项目：路由、Pydantic 校验、依赖注入、流式响应、容器化部署一条龙，所有代码均可直接复制运行，适合作为新服务的脚手架。
一、为什么选 FastAPI
在 Python Web 框架里，Django 偏重、Flask 偏薄，而 FastAPI 介于两者之间：它基于 Starlette（异步）和 Pydantic（数据校验），既轻量又类型安全。相比传统同步框架，FastAPI 用声明式类型自动完成参数解析、校验和文档生成，少写大量样板代码，同时在高并发 IO 场景下吞吐更高。
- 类型即文档：函数签名里的类型注解自动生成交互式 Swagger 文档。
- 异步原生：
async def视图直接跑在事件循环上，轻松应对成百上千并发连接。 - 校验内建：请求体、查询参数、路径参数都可由 Pydantic 模型约束，错误自动返回 422。
- 性能接近 Node/Go：底层是 ASGI，配合 uvloop 可逼近 Go 并发实战中 goroutine 的吞吐表现。
二、快速起步：安装与 Hello World
建议用独立虚拟环境隔离依赖。FastAPI 本体只负责路由与校验，真正运行还需一个 ASGI 服务器，官方推荐 Uvicorn。
# 创建并激活虚拟环境
python -m venv .venv
source .venv/bin/activate
# 安装核心依赖
pip install "fastapi[standard]"
# 最简应用 app.py
from fastapi import FastAPI
app = FastAPI(title="Demo API", version="1.0.0")
@app.get("/")
async def root():
return {"msg": "hello fastapi"}
# 本地启动
# uvicorn app:app --reload --host 0.0.0.0 --port 8000
访问 http://localhost:8000/docs 即可看到自动生成的 Swagger UI；/openapi.json 则输出标准 OpenAPI 描述，方便前端联调或接入网关。
三、路由与参数：路径、查询与校验
FastAPI 把路径参数、查询参数和请求体统一用类型注解声明，框架自动解析并做基础类型转换。
from typing import Annotated
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(
item_id: int, # 路径参数，自动转为 int
q: Annotated[str | None, Query(max_length=50)] = None, # 可选查询参数
page: Annotated[int, Query(ge=1)] = 1, # 带下限约束
):
return {"item_id": item_id, "q": q, "page": page}
当 item_id 无法转成整数，或 q 超过 50 字符，FastAPI 直接返回 422 并给出明确错误字段，无需手写校验逻辑。
四、请求体校验：Pydantic 模型
复杂入参用 Pydantic 模型描述，字段类型、默认值、约束一目了然，前端拿到的就是结构化数据。
from pydantic import BaseModel, Field
class CreateUser(BaseModel):
name: str = Field(..., min_length=1, max_length=30)
age: int = Field(..., ge=0, le=150)
email: str = Field(..., pattern=r"^[^@]+@[^@]+\.[^@]+$")
@app.post("/users")
async def create_user(u: CreateUser):
# u 已是校验通过的强类型对象
return {"id": 1, "name": u.name, "age": u.age}
| 字段约束写法 | 含义 |
|---|---|
Field(..., min_length=1) | 必填且最小长度 1 |
Field(..., ge=0, le=150) | 数值范围 0–150 |
Field(default=None) | 可选字段，默认 None |
pattern=r"..." | 正则校验字符串格式 |
五、异步视图与并发模型
FastAPI 的 async def 视图运行在单进程事件循环上。遇到数据库、HTTP 调用等 IO 等待时，用 await 让出控制权，事件循环转去处理其他请求，从而用少量线程扛住高并发。注意：CPU 密集任务要丢进线程池或进程池，否则会阻塞整个循环。
import asyncio
import httpx
from fastapi import FastAPI
app = FastAPI()
@app.get("/aggregate")
async def aggregate():
# 用异步客户端并发请求下游，整体耗时约等于最慢的一个
async with httpx.AsyncClient() as client:
r1, r2 = await asyncio.gather(
client.get("https://api.example.com/a"),
client.get("https://api.example.com/b"),
)
return {"a": r1.status_code, "b": r2.status_code}
若内部要调用同步阻塞库，用 def 声明视图，FastAPI 会自动把它扔到线程池执行，不阻塞异步请求。
六、依赖注入：复用鉴权与共享逻辑
Depends 是 FastAPI 的精髓：把”取当前用户””校验 Token””建数据库连接”等横切逻辑抽成依赖函数，在多个路由间复用，且自动参与 OpenAPI 文档。
from fastapi import Depends, HTTPException, Header
def get_token(x_token: Annotated[str, Header()]) -> str:
if x_token != "secret-demo":
raise HTTPException(status_code=401, detail="invalid token")
return x_token
@app.get("/protected")
async def protected(token: Annotated[str, Depends(get_token)]):
return {"status": "ok", "token": token[:4] + "****"}
七、中间件、CORS 与统一异常处理
跨域、日志、耗时统计等全局逻辑放到中间件；业务异常用自定义 ExceptionHandler 收敛成统一返回结构，前端无需逐接口适配。
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import time
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["https://fsdata.site"],
allow_methods=["*"],
allow_headers=["*"],
)
@app.middleware("http")
async def add_timing(request: Request, call_next):
start = time.perf_counter()
resp = await call_next(request)
resp.headers["X-Process-Time"] = f"{time.perf_counter()-start:.4f}"
return resp
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
return JSONResponse(status_code=400, content={"error": str(exc)})
八、流式响应：对接 SSE 长连接
做 AI 对话、日志推送时，可以用 StreamingResponse 把生成过程逐块吐给前端，体验类似打字机。关于 SSE 推流与前端的 EventSource 渲染细节，可参考我们此前的 LLM 流式输出实战：SSE 推流与前端渲染。
import asyncio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
async def token_stream(prompt: str):
for chunk in ["你好", "，", "这是", "流式", "返回"]:
yield chunk
await asyncio.sleep(0.2) # 模拟逐 token 生成
@app.get("/stream")
async def stream(prompt: str):
return StreamingResponse(token_stream(prompt), media_type="text/plain")
九、配置管理：pydantic-settings
把数据库地址、密钥等配置从代码里剥离，用环境变量注入，避免敏感信息进版本库。
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
db_url: str = "sqlite:///./app.db"
debug: bool = False
class Config:
env_file = ".env"
settings = Settings()
十、容器化部署与上线
生产环境用多阶段 Dockerfile 构建精简镜像，再用 Uvicorn 多 worker 启动。镜像构建与 Docker 编排 的最佳实践可一并参考，配合 GitHub Actions 实战：从零搭建 CI/CD 流水线 实现提交即构建发布。
# 多阶段构建，最终镜像仅含运行依赖
FROM python:3.12-slim AS builder
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
FROM python:3.12-slim
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
COPY app.py .
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
本地调试时在 VS Code 里配置 launch.json 直接断点 uvicorn，提升排错效率。
十一、生产落地检查清单
| 检查项 | 建议做法 |
|---|---|
| 进程模型 | Uvicorn 多 worker，或前置 Gunicorn + Uvicorn worker |
| 反向代理 | 前置 Nginx 做 TLS 终止与超时控制 |
| 超时与限流 | 网关层设请求超时，避免慢请求拖垮线程池 |
| 健康检查 | 暴露 /healthz 供 K8s 探针调用 |
| 日志 | 结构化 JSON 日志，接入集中采集 |
| 依赖锁 | 提交 requirements.txt 或 poetry.lock |
十二、三个常见踩坑
12.1 在 async 视图里调同步阻塞库
直接 requests.get() 会阻塞事件循环，导致所有并发请求卡顿。IO 场景一律改用 httpx.AsyncClient、asyncpg 等异步驱动；CPU 重活用 await run_in_threadpool(...) 或进程池。
12.2 Pydantic v1 / v2 混用
FastAPI 新版本默认 Pydantic v2，class Config 写法需改为 model_config = ConfigDict(...)。升级后务必跑一遍类型校验单测，避免运行时才暴露。
12.3 忘记关 CORS 与生产 Debug
allow_origins=["*"] 只适合联调，上线要收敛到具体域名；debug=True 会暴露堆栈与交互式调试器，生产务必关闭。
十三、后台任务：不阻塞响应的副作用
有些操作不必等它完成再返回，比如写审计日志、发通知、清理临时文件。用 BackgroundTasks 把这些工作挂到响应之后执行，既能立刻给用户返回结果，又不会丢任务。注意后台任务仍跑在同一进程里，重活应改投消息队列或异步 worker。
from fastapi import BackgroundTasks
def write_audit(user: str):
with open("audit.log", "a", encoding="utf-8") as f:
f.write(f"{user} called /protected\n")
@app.get("/protected")
async def protected(
token: Annotated[str, Depends(get_token)],
bg: BackgroundTasks,
):
bg.add_task(write_audit, token)
return {"status": "ok"}
十四、单元测试：用 TestClient 守住接口契约
FastAPI 基于 Starlette，自带 TestClient，可在不启动真实服务器的前提下同步调用接口，配合 pytest 写契约测试最省心。测试数据库建议用独立实例或事务回滚，避免污染线上数据。下面这段代码覆盖了正常返回与鉴权失败两条路径。
from fastapi.testclient import TestClient
def test_root():
client = TestClient(app)
assert client.get("/").json()["msg"] == "hello fastapi"
def test_protected_without_token():
client = TestClient(app)
# 未带 Token 应返回 401
assert client.get("/protected").status_code == 401
把测试接进 GitHub Actions 实战：从零搭建 CI/CD 流水线，每次提交自动跑 pytest，能在合并前拦住大部分破坏性改动，比上线后再救火成本低得多。
FastAPI 用类型把”校验、文档、并发”三件大事一次性解决，是 Python 后端 API 的高性价比选择。掌握路由、Pydantic、依赖注入与异步模型后，再补齐容器化、CI/CD 与单元测试，就能稳定地把服务送上生产环境。如果在高并发场景遇到瓶颈，也可对照 Go 并发实战：goroutine 与 channel 的思路做架构取舍。
