本节目标：用 FastAPI 从零写出可运行的 API，掌握路由、参数校验、响应模型、依赖与测试，理解
def与async def端点的执行差异。
适用版本：Python 3.12+（实测 3.14.6）
16.2 FastAPI 快速上手
16.1 节我们手写了 ASGI 应用，靠 await send({...}) 拼响应——能用，但太底层。这一节请出 FastAPI：它把 ASGI 的样板全部收进框架，让你用「类型注解 + 装饰器」描述接口，剩下的校验、序列化、文档全自动完成。安装（版本号来自本机实测）：pip install "fastapi==0.143.0" "uvicorn==0.51.0"。
16.2.1 FastAPI 的三块拼图
FastAPI 本身很薄，它站在两个库的肩膀上：
| 层 | 由谁提供 | 负责什么 |
|---|---|---|
| Web 框架 | Starlette（1.7.0） | 路由、中间件、ASGI 适配、TestClient |
| 数据校验 | Pydantic（2.13.5） | 请求/响应的模型校验与序列化 |
| 文档 | FastAPI 自研 | 由类型注解自动生成 OpenAPI + Swagger UI |
所以「会用 FastAPI」约等于「会用 Starlette 的路由 + Pydantic 的模型」——这两样我们在 9.3 节和 16.1 节都打过底。类型注解是唯一的真相来源：你写的注解同时驱动运行时校验和文档生成。
16.2.2 第一个应用
from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI(title="Book API", version="0.1.0")
@app.get("/")
def root():
return {"app": "Book API"}
@app.get("/items/{item_id}")
def get_item(item_id: int): # 路径参数，声明成 int
return {"item_id": item_id, "type": type(item_id).__name__}
client = TestClient(app)
print(client.get("/").json())
print(client.get("/items/42").json())
print(client.get("/items/abc").status_code)
真实输出：
{'app': 'Book API'}
{'item_id': 42, 'type': 'int'}
422
item_id: int 这行注解做了三件事：把路径里的字符串 "42" 转成整数 42、自动生成文档、非法时返回 422。用 uvicorn 起服务只需 uvicorn main:app --reload（main:app 指 main.py 里的 app 对象，--reload 改代码自动重启，仅开发用）。
16.2.3 自动文档：/docs 与 /redoc
FastAPI 启动时会扫描所有路由，生成一份 OpenAPI 规范，再据此渲染两套 UI。实测三个内置端点的状态码：
/docs 200 text/html # Swagger UI，可直接在页面里调接口
/redoc 200 text/html # ReDoc，更适合阅读
/openapi.json 200 application/json # 机器可读的规范本身
/openapi.json 里 /items/{item_id} 的真实片段（省略无关字段）：
{
"summary": "Get Item",
"parameters": [
{ "name": "item_id", "in": "path", "required": true,
"schema": { "type": "integer", "maximum": 1000, "minimum": 1 } }
],
"responses": { "200": { "description": "Successful Response" } }
}
注意 maximum: 1000, minimum: 1 是从 Path(ge=1, le=1000) 自动推出的——约束写一遍，运行时校验和文档同时生效。这份 openapi.json 还能直接喂给前端代码生成器，前后端联调从此少一半口水。
16.2.4 路径参数与查询参数的校验
用 Path / Query 给参数加约束，语义与 9.3 节的 Field 一致：
from fastapi import FastAPI, Path, Query
app = FastAPI()
@app.get("/items/{item_id}")
def get_item(item_id: int = Path(ge=1, le=1000)):
return {"item_id": item_id}
@app.get("/search")
def search(q: str = Query(min_length=2), limit: int = 10):
return {"q": q, "limit": limit}
GET /items/0 -> 422 (greater_than_equal)
GET /search?q=a -> 422 (string_too_short)
GET /search?q=hi -> 200 {"q": "hi", "limit": 10}
没有默认值的参数是必填的，有默认值就是可选。路径参数因为出现在 URL 模板里，永远必填。查询参数若写成 q: str（无默认值）也必填，访问不带 q 会 422。
16.2.5 请求体与 response_model
请求体用 Pydantic 模型声明，FastAPI 会自动解析 JSON 并校验：
from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel
app = FastAPI()
class UserIn(BaseModel):
name: str
email: str
password: str
class UserOut(BaseModel): # 出口模型：不含 password
name: str
email: str
@app.post("/users", response_model=UserOut)
def create_user(u: UserIn):
return u # 输入有 password，输出被过滤掉
c = TestClient(app)
print(c.post("/users", json={"name": "Ada", "email": "a@x.com", "password": "s3cr3t"}).json())
{'name': 'Ada', 'email': 'a@x.com'}
response_model 是一道安全闸门：它按 UserOut 重新序列化返回值，password 无论怎么混进来都出不去。永远不要让「输入模型」直接当「输出模型」。
想让「未显式设置的字段不出现在响应里」，加 response_model_exclude_unset=True：
class Profile(BaseModel):
name: str
bio: str = "这个人很懒"
website: str | None = None
@app.get("/profile", response_model=Profile, response_model_exclude_unset=True)
def profile():
return {"name": "Ada"} # bio/website 没设，就不输出
GET /profile -> {'name': 'Ada'}
不加这个开关，bio 会带上默认值一起返回。默认值适合「业务兜底」，exclude_unset 适合「PATCH 式部分更新」，按场景选。
16.2.6 状态码与 HTTPException
创建成功应回 201，用 status_code 声明；出错用 HTTPException：
from fastapi import FastAPI, HTTPException, status
from fastapi.testclient import TestClient
app = FastAPI()
ITEMS = {1: {"name": "键盘", "price": 199.0}}
@app.post("/items", status_code=status.HTTP_201_CREATED) # 创建成功回 201
def create_item():
return {"item_id": 3}
@app.get("/items/{item_id}")
def get_item(item_id: int):
if item_id not in ITEMS:
raise HTTPException(status_code=404, detail="Item not found")
return {"item_id": item_id, **ITEMS[item_id]}
c = TestClient(app)
print(c.post("/items").status_code)
print(c.get("/items/999").status_code, c.get("/items/999").json())
201
404 {'detail': 'Item not found'}
HTTPException 会被转成 {"detail": ...} 的 JSON 响应。注意区分：校验失败（422）是框架自动的，业务失败（404/409/…）要你自己 raise。
16.2.7 后台任务 BackgroundTasks
有些活儿不必让用户等（发邮件、写审计日志），丢给 BackgroundTasks：
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
log = []
def write_audit(msg: str):
log.append(msg)
@app.post("/signup")
def signup(bg: BackgroundTasks):
bg.add_task(write_audit, "audit: new signup")
return {"ok": True}
用 TestClient 请求后，响应体是 {'ok': True}，而 log 里已经多出一条 'audit: new signup'。
响应先返回，任务在响应之后执行。但它跑在同一个进程里，只适合几秒内的小任务；真正耗时的工作要用消息队列交给独立 worker，否则会拖慢甚至压垮 Web 进程。
16.2.8 路由分组：APIRouter
接口一多，路由堆在一个文件里就乱了。APIRouter 把一组路由装进一个对象，再用 prefix / tags 统一前缀与文档分组：
from fastapi import FastAPI, APIRouter
app = FastAPI()
router = APIRouter(prefix="/api/v1", tags=["v1"])
@router.get("/ping")
def ping():
return {"pong": True}
app.include_router(router)
此时 ping 的实际路径是 /api/v1/ping，且在 /docs 里归入 v1 分组。按业务拆成 users.py、orders.py 各自的 router，是 FastAPI 项目组织的标准做法。
16.2.9 Depends 的最小用法
依赖注入让「准备数据」和「处理请求」解耦。函数依赖写起来就是普通函数：
from typing import Annotated
from fastapi import FastAPI, Depends
app = FastAPI()
def pagination(skip: int = 0, limit: int = 10) -> dict:
return {"skip": skip, "limit": limit}
@app.get("/list")
def list_items(p: Annotated[dict, Depends(pagination)]):
return p
GET /list?skip=5&limit=20 -> {'skip': 5, 'limit': 20}
Depends(pagination) 告诉 FastAPI「先调用 pagination，把结果注入 p」。妙处在于 pagination 的 skip/limit 也是带类型的查询参数，校验与文档一并自动生成。带 yield 的依赖能做资源清理（等价于 pytest 的 fixture），16.3 节讲数据库时会重点展开。
16.2.10 async def 端点 vs def 端点
FastAPI 有个容易踩坑的行为：端点函数用 def 还是 async def，执行方式完全不同。
async def端点：在事件循环线程里直接运行。def端点：被丢进 anyio 线程池，由工作线程运行。
实测（uvicorn 起真实服务，打印各端点所在线程）：
uvicorn 事件循环线程 id = 6140866560
/async -> thread=6140866560 name=Thread-1 (run) 是事件循环线程: True
/sync -> thread=6157692928 name=AnyIO worker thread 是事件循环线程: False
含义很关键：def 端点里哪怕写了阻塞调用（requests、同步数据库），也只是占住一个工作线程，不会卡住整个事件循环；而 async def 端点里如果做了阻塞操作，会把整个事件循环卡死。实测 4 个并发请求各阻塞 0.5 秒：
4 个并发请求 /block-async（async def 里 time.sleep）耗时 2.05s
4 个并发请求 /block-sync （def 里 time.sleep） 耗时 0.52s
async def 里串行阻塞，4 个请求排成 2 秒；def 里被线程池并行处理，只要 0.5 秒。结论：async def 要配真正的异步库（如 13 章的 httpx.AsyncClient），def 才配同步库——最忌讳 async def 里塞同步阻塞代码。
16.2.11 用 TestClient 写测试
TestClient 让你不启动服务器就能测接口，写法像同步函数，底层走的是 ASGI：
from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
@app.get("/items/{item_id}")
def get_item(item_id: int):
return {"item_id": item_id}
client = TestClient(app)
def test_ok():
assert client.get("/items/1").json() == {"item_id": 1}
def test_bad_type():
assert client.get("/items/abc").status_code == 422
配合 14 章的 pytest 与 fixture，就能覆盖「正常路径 + 校验失败 + 业务异常」三类用例。response 有 .status_code / .json() / .headers，断言非常直接。
16.2.12 生产部署与多进程
开发用 uvicorn main:app --reload 单进程即可；生产用 uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 开多进程榨干多核。--workers 会拉起多个子进程，每个进程各跑一个事件循环。为什么要多进程而不是多线程？ 回到 12.1 节的 GIL：一个 CPython 进程同一时刻只有一个线程执行字节码，纯计算无法靠线程并行。ASGI 的 I/O 并发已经很强，但要吃满多核 CPU 仍得靠多进程。经验公式是 workers ≈ 2 × CPU 核数 + 1，但现代容器更推荐「一个容器一个进程」，用编排层横向扩容。
小结
- FastAPI = Starlette（Web）+ Pydantic（校验）+ 自动 OpenAPI 文档；类型注解是唯一真相来源。
- 路径参数靠注解自动转换与校验，
Path/Query加约束，非法输入统一 422。 response_model是出口安全闸门，response_model_exclude_unset控制部分更新；业务错误用HTTPException。BackgroundTasks只做轻量收尾，重活交给消息队列；APIRouter按业务拆分路由。Depends把准备数据与处理请求解耦；async def端点跑在事件循环线程，def端点跑在线程池——别在async def里写阻塞代码。- 生产用
uvicorn --workers多进程吃多核（GIL 决定了要进程而非线程）。
到这里，你已经能写出一批能跑的接口了。但真实服务要落库、要校验复杂规则、要复用连接——下一节我们把 Pydantic 校验、Depends 依赖注入和 SQLAlchemy 数据库访问接进来，拼成一个像样的后端。
阅读导航：上一节：HTTP 与 WSGI / ASGI · 下一节：数据校验、依赖注入与数据库访问 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。