10-测试与部署
Python 3.11+
本章讲解 FastAPI 应用测试和部署。
第一部分：测试基础
1.1 实际场景
需要验证 API 是否正常工作，测试各种输入情况。
问题：如何编写 API 自动化测试？
1.2 安装测试依赖
bash
pip install pytest httpx1.3 基本测试
python
from fastapi.testclient import TestClient
from main import app
client: TestClient = TestClient(app)
def test_root() -> None:
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello"}
def test_create_item() -> None:
response = client.post("/items", json={"name": "Test", "price": 10})
assert response.status_code == 201
data: dict = response.json()
assert data["name"] == "Test"第二部分：异步测试
2.1 实际场景
应用使用了异步操作，测试也需要异步执行。
问题：如何编写异步测试？
2.2 async 测试
python
import pytest
from httpx import AsyncClient
from main import app
@pytest.mark.asyncio
async def test_async() -> None:
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.get("/")
assert response.status_code == 200第三部分：部署
3.1 实际场景
开发完成后，需要将应用部署到生产服务器。
问题：如何部署 FastAPI 应用？
3.2 Uvicorn
bash
pip install uvicorn
# 运行
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 43.3 Gunicorn
bash
pip install gunicorn
# 运行
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker第四部分：Docker 部署
4.1 Dockerfile
dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]4.2 docker-compose.yml
yaml
version: '3.8'
services:
web:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://user:pass@db:5432/app
depends_on:
- db
db:
image: postgres:15
redis:
image: redis:7第五部分：完整示例
5.1 应用代码
python
# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
app: FastAPI = FastAPI()
# 模型
class Item(BaseModel):
id: Optional[int] = None
name: str
price: float
# 模拟数据库
items_db: list[Item] = []
# 路由
@app.get("/")
def root() -> dict[str, str]:
return {"message": "Hello API"}
@app.get("/items", response_model=list[Item])
def get_items() -> list[Item]:
return items_db
@app.post("/items", response_model=Item, status_code=201)
def create_item(item: Item) -> Item:
item.id = len(items_db) + 1
items_db.append(item)
return item
@app.get("/items/{item_id}", response_model=Item)
def get_item(item_id: int) -> Item:
for item in items_db:
if item.id == item_id:
return item
raise HTTPException(status_code=404, detail="Item not found")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)5.2 测试代码
python
# test_main.py
from fastapi.testclient import TestClient
from main import app, items_db
client: TestClient = TestClient(app)
def setup_function() -> None:
items_db.clear()
def test_root() -> None:
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello API"}
def test_create_item() -> None:
response = client.post("/items", json={"name": "Test", "price": 10})
assert response.status_code == 201
data: dict = response.json()
assert data["name"] == "Test"
assert data["price"] == 10
assert data["id"] == 1
def test_get_items() -> None:
client.post("/items", json={"name": "Item 1", "price": 5})
response = client.get("/items")
assert response.status_code == 200
assert len(response.json()) == 1
def test_get_item_not_found() -> None:
response = client.get("/items/999")
assert response.status_code == 4045.3 运行测试
bash
pytest test_main.py -v第六部分：L3 专家层
6.1 httpx TestClient 的 ASGI 直达（绕过网络层）
TestClient 的本质是一个 同步包装器，它将 HTTP 请求直接传递给 ASGI 应用，完全绕过网络栈。
请求流转对比：
# 真实 HTTP 请求
Client ──> TCP ──> HTTP Parser ──> ASGI Server ──> App
(socket) (解析协议) (Uvicorn)
# TestClient 请求
TestClient ──> ASGI Callable ──> App
(直接调用)TestClient 内部使用 httpx.Client，但将其传输层替换为 ASGITransport：
python
# httpx 内部逻辑示意（简化版）
class TestClient(httpx.Client):
def __init__(self, app: ASGIApp, ...) -> None:
transport = ASGITransport(app=app) # ◄── 关键：直达 ASGI
super().__init__(transport=transport, ...)
def get(self, url: str, **kwargs) -> Response:
# 不经过 TCP，直接构造 ASGI scope 并调用 app
# scope = {"type": "http", "method": "GET", "path": url, ...}
...ASGITransport 的工作方式：
TestClient.get("/items")
│
▼
构造 ASGI Scope
┌──────────────────────────────────┐
│ type: "http" │
│ method: "GET" │
│ path: "/items" │
│ headers: [(b"host", b"testserver")] │
│ query_string: b"" │
└───────────────┬──────────────────┘
│
▼
┌───────────────┐
│ app(scope, │ ◄── 直接调用 ASGI 应用
│ receive, │
│ send) │
└───────────────┘
│
▼
收集 send() 调用结果
┌──────────────────┐
│ response_start: │
│ status: 200 │
│ headers: ... │
│ │
│ response_body: │
│ body: {...} │
└────────┬─────────┘
│
▼
构造 httpx.Response 返回因为没有网络开销，TestClient 的测试速度极快（单次请求 < 1ms），但也意味着它 无法测试：
- 网络超时
- 连接池行为
- TLS/SSL 配置
- 真实反向代理的 header 改写
6.2 AsyncClient vs TestClient 的区别
| 维度 | TestClient | AsyncClient |
|---|---|---|
| API 风格 | 同步（client.get()） | 异步（await client.get()） |
| 事件循环 | 在测试线程中创建独立循环 | 复用 pytest-asyncio 提供的循环 |
| 底层实现 | httpx.Client + ASGITransport | httpx.AsyncClient + ASGITransport |
| 依赖注入 | 支持 | 支持 |
| websocket 测试 | 支持（with client.websocket_connect()） | 支持（async with client.websocket_connect()） |
| 适用场景 | 同步测试函数 | async def 测试函数 |
| 后台任务 | 在请求结束时同步执行 | 在请求结束时异步执行 |
核心区别——事件循环的处理：
TestClient（同步）
────────────────────────────────
测试线程 新线程
┌────────┐ ┌────────┐
│pytest │──调用──> │TestClient│
│测试函数│ │内部循环 │
│(同步) │<──响应── │await app │
└────────┘ └────────┘
AsyncClient（异步）
────────────────────────────────
同一个事件循环
┌──────────────────────────────────┐
│ pytest-asyncio loop │
│ │
│ async def test(): │
│ async with AsyncClient(): │
│ await client.get("/") │
│ └── await app(scope) │
└──────────────────────────────────┘选型建议：
你的 endpoint 是 async def 吗？
├── 是 ──> 优先使用 AsyncClient（避免 loop-in-loop 问题）
│
└── 否 ──> TestClient 即可
你的测试需要模拟真实异步行为吗？
├── 是 ──> AsyncClient
│
└── 否 ──> TestClient 更简洁6.3 Uvicorn 的 Worker 生命周期
Uvicorn 的进程模型基于 主进程 + Worker 进程 架构，理解其生命周期对部署和调试至关重要。
进程架构：
┌──────────────────────────────────────────────────────┐
│ Uvicorn Master │
│ │
│ ┌─────────────┐ │
│ │ 信号处理器 │ ◄── SIGTERM, SIGINT, SIGHUP │
│ └──────┬──────┘ │
│ │ │
│ ┌──────▼──────┐ fork/exec ┌──────────────────┐ │
│ │ Worker 管理器│ ──────────> │ Worker 1 │ │
│ │ │ │ (事件循环) │ │
│ │ --workers 4 │ fork/exec ├──────────────────┤ │
│ │ │ ──────────> │ Worker 2 │ │
│ │ │ │ (事件循环) │ │
│ │ │ fork/exec ├──────────────────┤ │
│ │ │ ──────────> │ Worker 3 │ │
│ │ │ │ (事件循环) │ │
│ │ │ fork/exec ├──────────────────┤ │
│ │ │ ──────────> │ Worker 4 │ │
│ │ │ │ (事件循环) │ │
│ └─────────────┘ └──────────────────┘ │
└──────────────────────────────────────────────────────┘Worker 生命周期状态机：
启动
│
▼
┌─────────────┐
│ BOOTING │ ◄── 导入应用、创建事件循环
└──────┬──────┘
│
▼
┌─────────────┐
│ STARTING │ ◄── 执行 startup lifespan
└──────┬──────┘
│
▼
┌─────────────┐
┌──────│ RUNNING │◄──────┐
│ └──────┬──────┘ │
│ │ │
│ 处理请求 │ │
│ │ │
▼ ▼ │
┌───────────┐ ┌───────────┐ │
│ GRACEFUL │ │ FORCED │ │
│ SHUTDOWN │ │ KILL │ │
│ │ │ │ │
│ 1. 停止 │ │ 立即终止 │ │
│ 接受新 │ │ │ │
│ 请求 │ │ │ │
│ 2. 等待 │ │ │ │
│ 现有 │ │ │ │
│ 请求 │ │ │ │
│ 完成 │ │ │ │
│ 3. 执行 │ │ │ │
│ shutdown││ │ │
│ lifespan││ │ │
│ 4. 退出 │ │ │ │
└─────┬─────┘ └─────┬─────┘ │
│ │ │
└─────────────┴──────────────┘
进程退出Lifespan 协议执行时机：
python
# FastAPI 应用的 lifespan 上下文
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
# ── STARTUP ──
# 在 Worker 进入 RUNNING 状态前执行
await db.connect()
yield
# ── SHUTDOWN ──
# 在 Worker 收到 SIGTERM 后、退出前执行
await db.disconnect()
app: FastAPI = FastAPI(lifespan=lifespan)关键生命周期参数：
| 参数 | 默认值 | 说明 |
|---|---|---|
--timeout-graceful-shutdown | 15s | 优雅关闭的等待时间，超时后强制终止 |
--workers | 1 | Worker 进程数量 |
--reload | False | 开发模式，文件变更时重启 Worker |
--backlog | 2048 | 等待处理的连接队列大小 |
优雅关闭的时间线：
T=0s 收到 SIGTERM
│
├── 停止 accept() 新连接
│
T=0.1s 发送 503 给新请求（如果有）
│
├── 等待现有请求完成
│
T=2s 请求 A 完成 ✓
T=5s 请求 B 完成 ✓
T=8s 所有请求完成
│
├── 执行 shutdown lifespan
│
T=8.1s 关闭数据库连接、清理资源
│
└── Worker 退出，Master 回收进程如果某个请求超过 --timeout-graceful-shutdown 仍未完成，Uvicorn 会强制终止该 Worker，可能导致数据不一致。因此长连接（如 WebSocket、SSE）需要特别处理。
总结
| 知识点 | 说明 |
|---|---|
| TestClient | 同步测试 |
| AsyncClient | 异步测试 |
| Uvicorn | ASGI 服务器 |
| Gunicorn | WSGI 服务器 |
| Docker | 容器化部署 |