第 10 章：FastAPI Web 框架

如果你熟悉 Express/Koa/Hono，FastAPI 会让你有宾至如归的感觉。它是 Python 目前最流行的现代 Web 框架，以高性能、自动 API 文档和类型安全著称。本章带你从零构建一个生产可用的 RESTful API。

10.1 为什么选 FastAPI

特性	FastAPI	Flask	Django REST
异步支持	✅ 原生 async/await	❌ 需扩展	⚠️ 部分支持
自动 API 文档	✅ Swagger + ReDoc	❌ 需手动	⚠️ 第三方
数据验证	✅ Pydantic 内置	❌ 需手动	✅ Serializer
性能	⭐ 极高 (Starlette)	⭐⭐ 高	⭐⭐⭐ 中
类型安全	✅ 完整类型提示	❌ 弱	⚠️ 部分
学习曲线	低（类 Express）	低	高

ℹ️前端类比

FastAPI 之于 Python 后端，就像 Next.js 之于 React 全栈——它结合了路由（Express）、数据验证（Zod/TypeScript）、API 文档（Swagger）于一身。如果你用过 Next.js 的 API Routes 或 tRPC，你会发现 FastAPI 的哲学与之如出一辙：用类型系统驱动开发。

10.2 安装与入门

bash

pip install "fastapi[standard]" uvicorn
# fastapi[standard] 会安装所有推荐依赖：pydantic, uvicorn, httpx 等

python

# main.py — 最小 FastAPI 应用
from fastapi import FastAPI

app = FastAPI(
    title="我的第一个 API",
    version="0.1.0"
)

@app.get("/")
async def root():
    return {"message": "Hello, FastAPI!"}

@app.get("/health")
async def health_check():
    return {"status": "ok"}

# 运行：uvicorn main:app --reload
# 访问：http://localhost:8000/docs（Swagger 文档）

💡uvicorn 是什么

uvicorn 是 ASGI（Async Server Gateway Interface）服务器，相当于 Python 界的 Node.js 运行时。它与 FastAPI 的关系就像 Node.js 与 Express——前者负责网络 I/O 和进程管理，后者负责路由和业务逻辑。

10.3 路由与请求处理

路径参数与查询参数

FastAPI 用 Python 类型提示自动完成参数校验和转换——你不需要写任何验证代码。

python

from fastapi import FastAPI
from typing import Optional

app = FastAPI()

# 路径参数（类似 Express 的 /users/:id）
@app.get("/users/{user_id}")
async def get_user(user_id: int):  # 自动转为 int，不符合 422 错误
    return {"user_id": user_id, "name": f"用户 {user_id}"}

# 查询参数（自动从 ?key=value 中提取）
@app.get("/search")
async def search(
    q: str,                    # 必填: /search?q=python
    page: int = 1,             # 可选，默认 1
    size: int = 20,            # 可选，默认 20
    sort: Optional[str] = None # 可选，可以为 None
):
    return {
        "query": q,
        "page": page,
        "size": size,
        "sort": sort,
        "results": [f"{q}_result_{page}_{i}" for i in range(size)]
    }

# 多路由方法
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"action": "read", "item_id": item_id}

@app.post("/items/{item_id}")
async def update_item(item_id: int):
    return {"action": "update", "item_id": item_id}

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    return {"action": "delete", "item_id": item_id}

10.4 Pydantic 模型——请求体验证

这是 FastAPI 最强大的特性之一。定义 Pydantic 模型来描述请求/响应的数据结构，FastAPI 自动完成：数据校验、类型转换、API 文档生成、JSON Schema 导出。

python

from fastapi import FastAPI
from pydantic import BaseModel, Field, EmailStr
from datetime import datetime
from typing import Optional

app = FastAPI()

# --- 请求模型（前端发送的数据） ---
class UserCreate(BaseModel):
    name: str = Field(min_length=2, max_length=50, examples=["张三"])
    email: EmailStr  # 内建邮箱验证
    age: int = Field(ge=0, le=150, description="年龄")
    bio: Optional[str] = Field(None, max_length=200)

    # 自定义验证（Pydantic v2）
    @classmethod
    def validate_name(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("姓名不能为空")
        return v.strip()

# --- 响应模型（返回给前端的数据） ---
class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    age: int
    bio: Optional[str] = None
    created_at: datetime

    # 控制序列化行为
    model_config = {
        "from_attributes": True  # 支持 ORM 对象
    }

# 模拟数据库
fake_db = []
counter = 0

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    """创建用户——请求体自动校验"""
    global counter
    counter += 1
    db_user = {
        "id": counter,
        "name": user.name,
        "email": user.email,
        "age": user.age,
        "bio": user.bio,
        "created_at": datetime.now()
    }
    fake_db.append(db_user)
    return db_user

@app.get("/users", response_model=list[UserResponse])
async def list_users(skip: int = 0, limit: int = 10):
    """用户列表"""
    return fake_db[skip: skip + limit]

ℹ️前端类比

Pydantic 模型 ≈ TypeScript 的 interface/type + Zod。当你定义 class UserCreate(BaseModel) 时，FastAPI 自动生成了等价的 JSON Schema 验证逻辑。这就像 tRPC 中你定义了一个 z.object({...}) schema，然后自动获得端到端的类型安全。

10.5 依赖注入

依赖注入（Dependency Injection）是 FastAPI 最独特的设计。它让你把可复用的逻辑（认证、数据库连接、权限检查）抽离成"依赖项"，通过路由参数自动注入。

python

from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Optional

app = FastAPI()

# ---- 可复用依赖 ----

# 1. 数据库连接（每个请求获取一次）
def get_db():
    """创建/关闭数据库连接——类似 Express 中间件中的 req.db"""
    db = {"connected": True}
    try:
        yield db
    finally:
        db["connected"] = False  # 请求结束后清理

# 2. 认证检查（依赖其他依赖）
async def get_current_user(
    authorization: str = Header(...),  # 必填 Header
    db: dict = Depends(get_db)         # 依赖其他依赖
):
    """从 Header 中提取 Token 并验证用户"""
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="缺少认证凭据")

    token = authorization.removeprefix("Bearer ")
    # 模拟验证：实际项目中应调用 JWT/数据库验证
    if token == "admin-token":
        return {"id": 1, "name": "管理员", "role": "admin"}
    else:
        return {"id": 2, "name": "普通用户", "role": "user"}

# 3. 权限检查（依赖认证结果）
def require_admin(user: dict = Depends(get_current_user)):
    """仅管理员可访问"""
    if user["role"] != "admin":
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return user

# ---- 路由中使用依赖 ----

@app.get("/me")
async def read_me(current_user: dict = Depends(get_current_user)):
    """任何已认证用户可访问"""
    return {"user": current_user}

@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    admin: dict = Depends(require_admin),  # 仅管理员
    db: dict = Depends(get_db)
):
    """需要管理员权限"""
    return {"message": f"用户 {user_id} 已删除", "by": admin["name"]}

ℹ️前端类比

FastAPI 的依赖注入就像 Express/Koa 的中间件链，但类型安全得多。你可以把它理解为 React 的 Context Provider——外层提供依赖，内层通过 Depends() 消费依赖。不同的是，FastAPI 的依赖可以嵌套、缓存、覆盖，且完全由类型系统驱动。

10.6 错误处理与响应

python

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()

# ---- 抛标准 HTTP 异常 ----
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id <= 0:
        raise HTTPException(
            status_code=422,
            detail={"field": "item_id", "message": f"无效 ID: {item_id}"}
        )
    if item_id > 100:
        raise HTTPException(status_code=404, detail="商品不存在")
    return {"id": item_id, "name": "Widget"}

# ---- 全局异常处理器 ----
class BusinessError(Exception):
    """自定义业务异常"""
    def __init__(self, code: str, message: str):
        self.code = code
        self.message = message

@app.exception_handler(BusinessError)
async def business_error_handler(request: Request, exc: BusinessError):
    """统一处理业务异常"""
    return JSONResponse(
        status_code=400,
        content={
            "error": exc.code,
            "message": exc.message,
            "path": str(request.url.path)
        }
    )

@app.get("/business/{action}")
async def business_route(action: str):
    if action == "fail":
        raise BusinessError(code="ACTION_DENIED", message="此操作不被允许")
    return {"action": action}

# ---- 自定义响应类 ----
class APIResponse(BaseModel):
    """统一响应格式"""
    code: int = 0
    message: str = "success"
    data: object = None

10.7 中间件与事件钩子

python

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
import time
import logging

app = FastAPI()

# ---- CORS 中间件（前端跨域必备） ----
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://myapp.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ---- 自定义中间件 ----
@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    """记录每个请求的耗时——类似 Express 的 morgan"""
    t0 = time.monotonic()
    response = await call_next(request)
    elapsed = time.monotonic() - t0
    response.headers["X-Response-Time"] = f"{elapsed:.4f}s"
    logging.info(f"{request.method} {request.url.path} -> {response.status_code} ({elapsed:.3f}s)")
    return response

# ---- 生命周期事件（启动/关闭钩子） ----
@app.on_event("startup")
async def startup():
    """应用启动时执行——初始化数据库连接池、缓存等"""
    print("🚀 服务启动中...")
    # await database.connect()

@app.on_event("shutdown")
async def shutdown():
    """应用关闭时执行——优雅关闭连接"""
    print("👋 服务正在关闭...")
    # await database.disconnect()

10.8 实战：RESTful API 完整示例

构建一个完整的「任务管理系统」API，涵盖 CRUD、认证、分页、搜索。

python

#!/usr/bin/env python3
"""
task_api.py
FastAPI 任务管理系统 — 完整 CRUD + 认证 + 分页
运行: uvicorn task_api:app --reload
文档: http://localhost:8000/docs
"""

from fastapi import FastAPI, HTTPException, Depends, Query, Header
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
import hashlib
import secrets

# ============================================================
# 应用初始化
# ============================================================
app = FastAPI(
    title="任务管理系统 API",
    description="一个完整的 FastAPI RESTful API 示例",
    version="1.0.0"
)

# ============================================================
# 数据模型
# ============================================================
class TaskCreate(BaseModel):
    """创建任务的请求体"""
    title: str = Field(min_length=1, max_length=100, examples=["学习 FastAPI"])
    description: Optional[str] = Field(None, max_length=500)
    priority: int = Field(default=1, ge=1, le=5, description="优先级 1-5")
    tags: list[str] = Field(default_factory=list, max_length=10)

class TaskUpdate(BaseModel):
    """更新任务——所有字段可选"""
    title: Optional[str] = Field(None, min_length=1, max_length=100)
    description: Optional[str] = Field(None, max_length=500)
    priority: Optional[int] = Field(None, ge=1, le=5)
    completed: Optional[bool] = None
    tags: Optional[list[str]] = None

class TaskResponse(BaseModel):
    """任务响应"""
    id: int
    title: str
    description: Optional[str]
    priority: int
    completed: bool
    tags: list[str]
    created_at: datetime
    updated_at: datetime

# ============================================================
# 模拟数据库
# ============================================================
tasks_db: dict[int, dict] = {}
task_counter = 0

def get_next_id() -> int:
    global task_counter
    task_counter += 1
    return task_counter

# ============================================================
# 认证依赖
# ============================================================
API_TOKENS = {"admin-token": "admin", "user-token": "user"}

def get_current_user(authorization: str = Header(...)):
    token = authorization.removeprefix("Bearer ")
    if token not in API_TOKENS:
        raise HTTPException(status_code=401, detail="无效的认证凭据")
    return {"username": API_TOKENS[token]}

# ============================================================
# API 路由
# ============================================================
@app.post("/tasks", response_model=TaskResponse, status_code=201)
async def create_task(
    task: TaskCreate,
    user: dict = Depends(get_current_user)
):
    """创建新任务"""
    task_id = get_next_id()
    now = datetime.now()
    task_dict = task.model_dump()
    task_dict.update(id=task_id, completed=False, created_at=now, updated_at=now)
    tasks_db[task_id] = task_dict
    return task_dict

@app.get("/tasks", response_model=list[TaskResponse])
async def list_tasks(
    page: int = Query(1, ge=1),
    size: int = Query(20, ge=1, le=100),
    q: Optional[str] = None,
    completed: Optional[bool] = None,
    user: dict = Depends(get_current_user)
):
    """任务列表——支持分页、搜索、过滤"""
    results = list(tasks_db.values())

    # 搜索（标题或描述包含关键词）
    if q:
        q_lower = q.lower()
        results = [t for t in results
                   if q_lower in t["title"].lower()
                   or (t.get("description") and q_lower in t["description"].lower())]

    # 状态过滤
    if completed is not None:
        results = [t for t in results if t["completed"] == completed]

    # 按创建时间倒序
    results.sort(key=lambda t: t["created_at"], reverse=True)

    # 分页
    start = (page - 1) * size
    return results[start: start + size]

@app.get("/tasks/{task_id}", response_model=TaskResponse)
async def get_task(
    task_id: int,
    user: dict = Depends(get_current_user)
):
    """获取单个任务"""
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail=f"任务 {task_id} 不存在")
    return tasks_db[task_id]

@app.put("/tasks/{task_id}", response_model=TaskResponse)
async def update_task(
    task_id: int,
    update: TaskUpdate,
    user: dict = Depends(get_current_user)
):
    """更新任务——仅更新提供的字段"""
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail=f"任务 {task_id} 不存在")

    # 只更新非 None 的字段（部分更新/PATCH 语义在这里用 PUT 演示）
    update_data = update.model_dump(exclude_unset=True)
    tasks_db[task_id].update(update_data)
    tasks_db[task_id]["updated_at"] = datetime.now()

    return tasks_db[task_id]

@app.delete("/tasks/{task_id}", status_code=204)
async def delete_task(
    task_id: int,
    user: dict = Depends(get_current_user)
):
    """删除任务"""
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail=f"任务 {task_id} 不存在")
    del tasks_db[task_id]
    return None  # 204 No Content


# ============================================================
# 启动说明
# ============================================================
if __name__ == "__main__":
    import uvicorn
    uvicorn.run("task_api:app", host="0.0.0.0", port=8000, reload=True)

10.9 FastAPI 进阶主题

后台任务

适合在请求返回后执行耗时操作（发邮件、生成报告），不阻塞客户端响应。

python

from fastapi import BackgroundTasks

def send_welcome_email(email: str, username: str):
    """模拟发送邮件（耗时 5 秒）"""
    import time
    time.sleep(5)  # 模拟
    print(f"已发送欢迎邮件到 {email}")

@app.post("/register")
async def register(email: str, background_tasks: BackgroundTasks):
    # 把耗时任务加入后台队列
    background_tasks.add_task(send_welcome_email, email, "新用户")
    # 立即返回，不等待邮件发送完成
    return {"message": "注册成功，确认邮件稍后将发送到您的邮箱"}

WebSocket 支持

FastAPI 原生支持 WebSocket——这对前端开发者来说再熟悉不过了。

python

from fastapi import WebSocket, WebSocketDisconnect

class ConnectionManager:
    """WebSocket 连接管理器——类似 Socket.IO 的 rooms"""
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active_connections.append(ws)

    def disconnect(self, ws: WebSocket):
        self.active_connections.remove(ws)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(ws: WebSocket, client_id: str):
    await manager.connect(ws)
    try:
        while True:
            data = await ws.receive_text()
            await manager.broadcast(f"客户端 {client_id}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(ws)
        await manager.broadcast(f"客户端 {client_id} 已离开")

文件上传

python

from fastapi import UploadFile, File
from pathlib import Path

UPLOAD_DIR = Path("./uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    # 安全检查：限制文件大小和类型
    MAX_SIZE = 10 * 1024 * 1024  # 10MB

    content = await file.read()
    if len(content) > MAX_SIZE:
        raise HTTPException(status_code=413, detail="文件过大")

    # 保存文件
    file_path = UPLOAD_DIR / file.filename
    with open(file_path, "wb") as f:
        f.write(content)

    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(content),
        "path": str(file_path)
    }

10.10 测试 FastAPI 应用

FastAPI 提供了 TestClient，让你用 pytest 编写端到端 API 测试——不需要启动真实服务器。

bash

pip install httpx pytest

python

# test_api.py
from fastapi.testclient import TestClient
from task_api import app

client = TestClient(app)

# 覆盖认证依赖（测试时跳过真实认证）
def override_get_current_user():
    return {"username": "test-user"}

from task_api import get_current_user
app.dependency_overrides[get_current_user] = override_get_current_user

class TestTaskAPI:
    def test_create_task(self):
        resp = client.post("/tasks", json={
            "title": "测试任务",
            "priority": 3,
            "tags": ["test"]
        }, headers={"Authorization": "Bearer test-token"})
        assert resp.status_code == 201
        data = resp.json()
        assert data["title"] == "测试任务"
        assert data["completed"] == False

    def test_list_tasks(self):
        resp = client.get("/tasks", headers={"Authorization": "Bearer test-token"})
        assert resp.status_code == 200
        assert isinstance(resp.json(), list)

    def test_get_nonexistent_task(self):
        resp = client.get("/tasks/99999", headers={"Authorization": "Bearer test-token"})
        assert resp.status_code == 404

    def test_create_task_invalid(self):
        resp = client.post("/tasks", json={
            "title": "",          # 空标题
            "priority": 10        # 超出范围
        }, headers={"Authorization": "Bearer test-token"})
        assert resp.status_code == 422  # 自动验证错误

    def test_unauthorized(self):
        resp = client.get("/tasks")  # 不传 Authorization
        assert resp.status_code == 401

10.11 部署建议

生产服务器：使用 gunicorn（Linux）管理多个 uvicorn worker 进程，前面加 Nginx 反向代理。
Docker：建议将所有依赖冻结到 requirements.txt，用多阶段构建减小镜像体积。
环境变量：敏感配置（API Key、数据库密码）通过环境变量注入，不要硬编码。
日志：生产环境使用 JSON 格式的结构化日志（如 python-json-logger），便于日志聚合系统采集。

dockerfile

# Dockerfile 示例
FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# 多 worker 生产启动
CMD ["uvicorn", "task_api:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

10.12 本章小结

📝要点回顾

FastAPI 是 Python 最快的现代 Web 框架，原生支持 async/await
Pydantic 模型提供自动数据验证，等价于 TypeScript + Zod
依赖注入系统让认证、数据库连接等逻辑高度复用
uvicorn 是 ASGI 服务器，负责网络 I/O 和进程管理
/docs 自动生成 Swagger 交互式文档
TestClient 让你不启动服务器就能端到端测试 API
后台任务和 WebSocket 支持让 FastAPI 可以胜任实时应用

🚀下一步

恭喜你完成了全部 10 章的学习！建议接下来：选择一个实际项目（如个人博客 API、数据分析工具、AI 聊天助手），将各章节的知识串联起来实践。编程是一门手艺，只有持续写代码才能真正内化这些知识。