FastAPI：从类型契约到 HTTP API
目标：能够把已经验证过的 Pydantic 模型和异步服务暴露为 FastAPI endpoint，并用 TestClient 验证成功与失败路径。
FastAPI 根据函数签名判断数据来源：路径模板中的参数来自 path，简单标量通常来自 query，Pydantic model 通常来自 request body。它把 Pydantic JSON Schema 放入 OpenAPI，并提供交互文档。
async def 不是“更快”的装饰：当 endpoint 调用支持 await 的数据库或 HTTP client 时使用 async；如果第三方库只有阻塞同步 API，普通 def endpoint 会由 FastAPI 放到线程池。你在 async endpoint 中直接调用的普通工具函数不会被自动转线程。
安装与应用
运行：
uv add "fastapi[standard]"
uv add --dev httpx
建立 src/textlab/api.py：
from __future__ import annotations
import asyncio
from fastapi import FastAPI, HTTPException
from textlab.schemas import MessageInput, MessageOutput
app = FastAPI(title="TextLab", version="1.0.0")
async def classify_async(message: MessageInput) -> MessageOutput:
await asyncio.sleep(0)
category = "refund" if "退款" in message.text else "general"
return MessageOutput(
text=message.text,
category=category,
priority=message.priority,
)
@app.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok"}
@app.post("/messages", response_model=MessageOutput)
async def create_message(message: MessageInput) -> MessageOutput:
try:
async with asyncio.timeout(2):
return await classify_async(message)
except TimeoutError as exc:
raise HTTPException(status_code=504, detail="processing timed out") from exc
启动：
uv run fastapi dev src/textlab/api.py
预期观察：终端显示服务地址，打开 http://127.0.0.1:8000/docs 可以看到 /health 和 /messages。请求：
curl -s http://127.0.0.1:8000/messages \
-H 'content-type: application/json' \
-d '{"text":"我要退款","priority":2,"metadata":{"source":"curl"}}'
预期 JSON：
{"text":"我要退款","category":"refund","priority":2}
HTTP 层应该保持薄
route 负责：接收/返回模型、认证/依赖、HTTP 状态映射。业务分类和并发 orchestration 应在可独立测试的 service 中。当前 classify_async 仍放在同一文件是为了先观察框架；Capstone 会把它移动到 service。
不要用 except Exception 把所有失败都映射成 500。已知输入错误由 Pydantic 产生 422，已知领域错误和 timeout 应有明确映射，未知错误让全局处理与日志系统记录。
引导实验：用 TestClient 固定契约
建立 tests/test_api.py：
from fastapi.testclient import TestClient
from textlab.api import app
client = TestClient(app)
def test_health() -> None:
response = client.get("/health")
assert response.status_code == 200
assert response.json() == {"status": "ok"}
def test_create_message() -> None:
response = client.post(
"/messages",
json={"text": "我要退款", "priority": 2, "metadata": {"source": "test"}},
)
assert response.status_code == 200
assert response.json()["category"] == "refund"
def test_rejects_unknown_field() -> None:
response = client.post(
"/messages",
json={
"text": "hello",
"metadata": {"source": "test"},
"pritorty": 3,
},
)
assert response.status_code == 422
运行：
uv run pytest tests/test_api.py -q
预期输出：3 passed。TestClient 调用是普通同步调用；若测试还要直接 await async database function，才需要 async test 工具。
独立练习
新增 GET /messages/{message_id} 的内存实现：不存在返回 404。为避免测试顺序污染，用 FastAPI dependency 或 app factory 注入一个新 dict，而不是永久 module global。
参考设计
先写纯 repository 类或 dict wrapper，测试中创建新的 app/repository。最小验收：
def test_missing_message_returns_404() -> None:
response = client.get("/messages/missing")
assert response.status_code == 404
assert response.json() == {"detail": "message not found"}
若选择暂时使用 module dict，至少提供 pytest fixture 在每个测试前清空；第 17 课会用数据库替换。