中间件、CORS 与全局异常

第 8 章 · 中间件、CORS 与全局异常

本章目标：理解 FastAPI/Starlette 中间件执行顺序；配置 CORSMiddleware 支持前端跨域；使用 @app.exception_handler 捕获业务异常与校验错误；设计 统一错误响应 JSON 结构；为 svc-demo 接入可观测的请求 ID 与访问日志。

学时建议：4～5 小时（含 1.5 小时跨域与异常跟练）

前置：完成 fastapi-web ch07（JWT、get_current_user）；ch04 响应模型基础。

8.1 请求生命周期与中间件位置

客户端请求
  → Server（Uvicorn）
  → 外层中间件（先注册的后执行「进入」）
  → 内层中间件
  → 路由匹配 → 依赖注入 → 路由函数
  → 响应
  → 中间件「离开」顺序相反
  → 返回客户端

层次	典型职责
最外层	请求 ID、访问日志、超时
CORS	预检 OPTIONS、响应头
认证（也可用依赖）	部分路径全局校验
业务路由	CRUD、业务异常

svc-demo 本地 http://127.0.0.1:8000；Vue/React 开发服 http://localhost:5173 需 CORS。生产前端 https://www.example.com 调 https://api.example.com 同理。

8.2 CORSMiddleware 配置

app/main.py：

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.core.config import settings

app = FastAPI(title="svc-demo API", version="1.0.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["X-Request-ID"],
)

app/core/config.py：

class Settings(BaseSettings):
    # ...
    CORS_ORIGINS: list[str] = [
        "http://localhost:5173",
        "http://127.0.0.1:5173",
        "https://www.example.com",
    ]

参数	说明
`allow_origins`	白名单；生产勿用 `["*"]` + `credentials=True`
`allow_credentials`	允许 Cookie / Authorization（需指定源）
`allow_methods`	常用 `GET,POST,PUT,PATCH,DELETE,OPTIONS`
`expose_headers`	前端可读自定义响应头

预检请求：浏览器对跨域非简单请求先发 OPTIONS，CORSMiddleware 自动响应，无需手写路由。

curl 模拟预检：

curl -i -X OPTIONS http://127.0.0.1:8000/api/v1/products \
  -H "Origin: http://localhost:5173" \
  -H "Access-Control-Request-Method: POST"

响应应含 access-control-allow-origin。

8.3 请求 ID 中间件

RequestIDMiddleware：从请求头读取或生成 X-Request-ID，写入 request.state 并附加到响应头，便于日志关联（ch10 Celery payload 可透传）。

class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4())
        request.state.request_id = request_id
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        return response

注册顺序见 8.8；访问日志可在同一中间件记录 method path status duration_ms。

8.4 统一错误响应结构

对外 API（含 api.example.com）建议固定错误 JSON，便于前端统一处理：

{
  "success": false,
  "error": {
    "code": "PRODUCT_NOT_FOUND",
    "message": "商品不存在",
    "details": null
  },
  "request_id": "550e8400-e29b-41d4-a716-446655440000"
}

app/schemas/errors.py：

from typing import Any

from pydantic import BaseModel


class ErrorBody(BaseModel):
    code: str
    message: str
    details: Any | None = None


class ErrorResponse(BaseModel):
    success: bool = False
    error: ErrorBody
    request_id: str | None = None

app/core/exceptions.py：

class AppException(Exception):
    def __init__(
        self,
        code: str,
        message: str,
        status_code: int = 400,
        details: dict | list | None = None,
    ):
        self.code = code
        self.message = message
        self.status_code = status_code
        self.details = details


class NotFoundError(AppException):
    def __init__(self, message: str = "资源不存在", code: str = "NOT_FOUND"):
        super().__init__(code=code, message=message, status_code=404)


class ConflictError(AppException):
    def __init__(self, message: str, code: str = "CONFLICT"):
        super().__init__(code=code, message=message, status_code=409)


class UnauthorizedError(AppException):
    def __init__(self, message: str = "未授权", code: str = "UNAUTHORIZED"):
        super().__init__(code=code, message=message, status_code=401)

8.5 注册 exception_handler