FastAPI 中间件详解：从原理、注册到异常捕获与实战应用-js教程-PHP中文网

FastAPI 中间件详解：从原理、注册到异常捕获与实战应用

本文系统讲解 FastAPI 中间件的核心机制，涵盖洋葱模型执行流程、@app.middleware("http") 与 add_middleware() 两种注册方式、耗时统计/请求拦截/响应头注入等典型用法，并重点说明全局异常处理与 CORS 配置的避坑要点。

本文系统讲解 fastapi 中间件的核心机制，涵盖洋葱模型执行流程、`@app.middleware("http")` 与 `add_middleware()` 两种注册方式、耗时统计/请求拦截/响应头注入等典型用法，并重点说明全局异常处理与 cors 配置的避坑要点。

FastAPI 中间件（Middleware）是位于客户端请求与服务端响应之间的“逻辑夹层”，采用经典的洋葱模型（Onion Model） 执行：请求由外向内逐层穿透中间件，抵达路由函数后，响应再由内向外逐层返回。这种双向可插拔的设计，使开发者能在不侵入业务代码的前提下，统一实现鉴权、日志、限流、跨域、性能监控与异常兜底等关键能力。

一、中间件的本质与执行流程

中间件本质上是一个异步函数，接收两个参数：request: Request（当前请求对象）和 call_next: Callable（继续向下传递的钩子）。其核心在于对 call_next(request) 的调用时机：

✅ 不调用 call_next → 立即终止流程，提前返回（如 Token 校验失败返回 401）；
✅ 调用 call_next 后处理 response → 在响应返回前注入逻辑（如添加 X-Process-Time 头）；
✅ 在 call_next 前后均操作 → 实现完整生命周期控制（如计时 + 日志）。

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import time

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    # 【请求阶段】—— 路由执行前
    start_time = time.perf_counter()

    try:
        # 执行后续中间件 + 路由函数
        response = await call_next(request)
    except Exception as exc:
        # 【异常兜底】—— 可在此捕获未被路由处理器处理的异常
        return JSONResponse(
            status_code=500,
            content={"code": 500, "message": "服务器内部错误", "data": None}
        )

    # 【响应阶段】—— 路由执行后
    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = f"{process_time:.3f}s"
    return response

⚠️ 注意：@app.middleware("http") 注册方式简洁，但仅适用于单文件快速原型；中大型项目推荐使用 BaseHTTPMiddleware 子类 + add_middleware() 方式，便于模块化管理与单元测试。

二、结构化注册：基于 `BaseHTTPMiddleware` 的专业实践

创建自定义中间件类需继承 starlette.middleware.base.BaseHTTPMiddleware，并重写 dispatch 方法：

# app/middleware/usetime_middleware.py
import time
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

class UseTimeMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        start_time = time.perf_counter()
        response = await call_next(request)
        process_time = time.perf_counter() - start_time
        response.headers["X-Process-Time"] = str(process_time)
        return response

在应用初始化时统一注册：

# app/main.py
from fastapi import FastAPI
from app.middleware.usetime_middleware import UseTimeMiddleware

app = FastAPI()

# ✅ 推荐：显式、可配置、易维护
app.add_middleware(UseTimeMiddleware)

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

三、全局异常捕获：不止于 `HTTPException`

FastAPI 默认仅捕获 HTTPException 和 RequestValidationError，但生产环境需覆盖所有未处理异常（如数据库连接失败、第三方 API 超时）。推荐两种方案：

Fastapi Code Review

审查 FastAPI 代码的路由模式、依赖注入、验证和异步处理器。适用于审查 FastAPI 应用、检查 APIRouter 配置、依赖注入等。

下载

方案1：中间件兜底（最简可靠）

@app.middleware("http")
async def global_exception_handler(request: Request, call_next):
    try:
        return await call_next(request)
    except Exception as exc:
        # 记录详细错误日志（建议接入 Sentry/Prometheus）
        print(f"[ERROR] Unhandled exception: {exc}")
        return JSONResponse(
            status_code=500,
            content={"code": 500, "message": "服务暂时不可用", "data": None}
        )

方案2：组合式异常处理器（更精细）

from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={"code": exc.status_code, "message": exc.detail, "data": None}
    )

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    errors = "; ".join([f"{'.'.join(e['loc'])}: {e['msg']}" for e in exc.errors()])
    return JSONResponse(
        status_code=422,
        content={"code": 422, "message": f"参数校验失败: {errors}", "data": None}
    )

✅ 提示：exception_handler 优先级高于中间件，适合按异常类型定制响应；中间件则更适合统一日志、指标埋点等横切关注点。

四、CORS 配置避坑指南（高频故障点）

FastAPI 使用 CORSMiddleware 处理跨域，但极易因配置不当导致前端静默失败（浏览器预检 OPTIONS 请求被拒）：

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://your-frontend.com"],  # ❌ 禁止用 ["*"] + allow_credentials=True
    allow_credentials=True,                        # ✅ 允许 Cookie/Authorization
    allow_methods=["*"],
    allow_headers=["*"],                           # ✅ 显式放行 Authorization, Content-Type
    expose_headers=["X-Process-Time"]              # ✅ 若前端需读取自定义响应头
)

⚠️ 关键约束：

allow_origins=["*"] 与 allow_credentials=True 互斥，否则中间件失效；
前端若携带 Authorization 或 Content-Type: application/json，必须确保 allow_headers 包含对应值；
源地址必须协议、域名、端口完全匹配（http://localhost:3000 ≠ https://localhost:3000）。

总结

FastAPI 中间件是构建健壮 Web 服务的基石能力。掌握其洋葱模型本质、区分 @middleware 与 add_middleware 的适用场景、熟练编写耗时统计/鉴权拦截/异常兜底中间件，并规避 CORS 配置陷阱，即可大幅提升开发效率与系统可观测性。建议将通用中间件（如日志、熔断、追踪）沉淀为可复用包，配合 OpenTelemetry 实现全链路监控，让 API 服务真正具备生产就绪（Production-Ready）能力。