fastapi

FastAPI 核心知识

完整参考：查看 advanced.md，了解 WebSocket 集成，包括连接管理、身份验证、房间管理、Pydantic 消息协议、心跳、Redis pub/sub 扩展和后台任务。

深入了解：使用 mcp__documentation__fetch_docs，并将技术设置为 fastapi，获取完整文档。

基本设置

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, EmailStr

app = FastAPI(title="My API")

class UserCreate(BaseModel):
    name: str
    email: EmailStr

class User(UserCreate):
    id: int

    class Config:
        from_attributes = True

路由模式

@app.get("/users", response_model=list[User])
async def get_users(skip: int = 0, limit: int = 100):
    return await db.users.find_many(skip=skip, limit=limit)

@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: int):
    user = await db.users.find(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.post("/users", response_model=User, status_code=201)
async def create_user(user: UserCreate):
    return await db.users.create(user.model_dump())

依赖注入

async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = await verify_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

@app.get("/me", response_model=User)
async def get_me(user: User = Depends(get_current_user)):
    return user

主要特性

• 在 /docs 自动生成 OpenAPI/Swagger 文档
• Pydantic 验证
• 支持异步
• 全面使用类型提示

不适合使用此技能的情况

• Django 项目——Django 有自己的 ORM、管理后台和模板
• Flask 微服务——Flask 更轻量，不会带来类型验证开销
• 同步 WSGI 应用——FastAPI 以异步为先
• 旧版 Python 2.x——FastAPI 要求 Python 3.7+
• 非 REST API——使用专门的 GraphQL 或 gRPC 框架

反模式

反模式	问题所在	解决方案
使用 def 而不是 async def	阻塞事件循环	对 I/O 操作使用 async def
缺少 response_model	不会验证输出	始终指定 response_model
同步数据库调用	阻塞工作进程	使用异步驱动程序（asyncpg、motor）
使用未加锁的全局状态	产生竞态条件	使用 asyncio.Lock 或 Depends()
引发异常而不使用 HTTPException	返回通用的 500 错误	使用带状态码的 HTTPException
不验证输入	存在安全漏洞	使用带验证器的 Pydantic 模型

快速故障排查

问题	诊断	修复方法
"RuntimeError: no running event loop"	在同步代码中调用异步代码	使用 await 或 asyncio.run()
验证错误不够明确	缺少字段描述	添加 Field(description=...)
响应时间慢	同步数据库调用	切换到异步 SQLAlchemy
浏览器中出现 CORS 错误	缺少中间件	添加 CORSMiddleware
未注入依赖	导入或语法错误	检查 Depends() 语法

生产就绪

安全配置

from fastapi.middleware.cors import CORSMiddleware
from slowapi import Limiter

app = FastAPI(
    title="My API",
    docs_url="/docs" if os.getenv("ENV") != "production" else None,
)

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

app.add_middleware(
    CORSMiddleware,
    allow_origins=os.getenv("ALLOWED_ORIGINS", "").split(","),
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
    allow_headers=["*"],
)

健康检查

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

@app.get("/ready")
async def readiness(db: Session = Depends(get_db)):
    try:
        db.execute("SELECT 1")
        return {"status": "ready", "database": "connected"}
    except Exception:
        return JSONResponse(status_code=503, content={"status": "not ready"})

优雅关停

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    await database.connect()
    yield
    await database.disconnect()

app = FastAPI(lifespan=lifespan)

监控指标

指标	告警阈值
请求延迟 p99	> 500ms
错误率（5xx）	> 1%
内存使用率	> 80%
工作进程利用率	> 90%

检查清单

• 正确配置 CORS
• 启用速率限制
• 配置安全标头中间件
• 对所有输入进行 Pydantic 验证
• 配置健康/就绪/存活端点
• 使用结构化日志（JSON 格式）
• 配置全局异常处理器
• 通过环境变量管理密钥
• 在生产环境中禁用文档
• 使用 Gunicorn 并配置多个工作进程
• 处理优雅关停

参考文档

• Pydantic 模型
• 依赖项