FastAPI中间件完全指南

📂 所属阶段：第二阶段 — 进阶黑科技（核心篇）
🔗 相关章节：FastAPI异步编程深度解析 · FastAPI异常处理

目录

一、中间件基础入门

什么是中间件？

中间件是包裹整个应用的代码层：请求到达路由前先经过它，路由返回响应后再反向经过它。你可以把它理解为请求的「安检通道」和「包装工厂」。

请求生命周期中的位置

sequenceDiagram
    participant Client as 客户端
    participant M1 as 中间件1(日志)
    participant M2 as 中间件2(CORS)
    participant M3 as 中间件3(GZip)
    participant Route as 路由处理函数
    Client->>M1: 请求
    M1->>M2: 请求
    M2->>M3: 请求
    M3->>Route: 请求
    Route-->>M3: 响应
    M3-->>M2: 压缩后的响应
    M2-->>M1: CORS响应
    M1-->>Client: 带日志的响应

中间件 vs 依赖注入

FastAPI中「中间件」和「依赖注入」是两种互补的横切关注点处理方式，区别如下：

特性中间件依赖注入
触发范围所有请求自动经过按需注入到路由参数
执行逻辑先注册的先处理请求、后处理响应与路由参数顺序一致
用途全局拦截、压缩、CORS、全局日志参数提取、单路由认证、数据库连接
响应控制可以提前返回/修改响应不能直接返回响应

核心优势

  1. 全局统一：避免在每个路由重复写相同逻辑
  2. 关注点分离：将日志、安全等与业务解耦
  3. 灵活复用：同一中间件可应用到多个项目
  4. 性能友好：内置异步支持，不阻塞请求

二、中间件开发基础

FastAPI支持两种中间件实现方式：

方式1：装饰器中间件（简单快速）

适合无初始化参数、无内部状态的轻量功能，比如耗时统计。

from fastapi import FastAPI, Request
import time

app = FastAPI()

@app.middleware("http")
async def add_process_time(request: Request, call_next):
    # ── 请求前处理 ──
    start = time.perf_counter()

    # ── 传递请求给下一层 ──
    response = await call_next(request)

    # ── 响应后处理 ──
    cost = (time.perf_counter() - start) * 1000
    response.headers["X-Process-Time"] = f"{cost:.2f}ms"

    return response

方式2：继承BaseHTTPMiddleware（复杂功能）

适合需要初始化参数（比如配置过滤路径、日志等级）或内部状态（比如性能统计缓存）的中间件。

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SimpleLogMiddleware(BaseHTTPMiddleware):
    """带初始化参数的简单日志中间件"""
    def __init__(self, app, skip_paths: list = None):
        super().__init__(app)
        self.skip_paths = skip_paths or ["/health", "/docs", "/redoc"]

    async def dispatch(self, request: Request, call_next) -> Response:
        # 跳过指定路径
        if request.url.path in self.skip_paths:
            return await call_next(request)

        # 记录请求
        logger.info(f"📥 {request.method} {request.url.path}")
        response = await call_next(request)
        # 记录响应
        logger.info(f"📤 {request.method} {request.url.path} → {response.status_code}")
        return response

# 注册中间件
app.add_middleware(SimpleLogMiddleware, skip_paths=["/metrics", "/docs"])

三、常用官方/内置中间件

1. CORS跨域中间件

FastAPI内置了官方维护的CORSMiddleware，解决浏览器跨域问题。

生产环境推荐配置

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    # ✅ 明确指定允许的域名，不要用 *
    allow_origins=[
        "https://daomanpy.com",
        "https://www.daomanpy.com",
        "http://localhost:3000",  # 开发环境临时加
    ],
    # ✅ 允许携带 Cookie（必须配合明确的 allow_origins）
    allow_credentials=True,
    # ✅ 限制允许的 HTTP 方法
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    # ✅ 限制允许的请求头
    allow_headers=["Content-Type", "Authorization", "X-Request-ID"],
    # ✅ 暴露给前端的响应头
    expose_headers=["X-Process-Time", "X-Request-ID"],
    # ✅ 预检请求缓存 1 小时，减少重复请求
    max_age=3600,
)

2. GZip响应压缩中间件

Starlette内置的GZipMiddleware，自动压缩文本类响应（如JSON、HTML、JS），减少网络传输量。

from starlette.middleware.gzip import GZipMiddleware

# 仅压缩大于 1000 字节的响应
app.add_middleware(GZipMiddleware, minimum_size=1000)

四、高级自定义中间件实现

1. 安全头中间件

给所有响应添加OWASP推荐的安全头部，提升应用安全性。

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        response = await call_next(request)
        # 添加安全头部
        response.headers.update({
            "X-Content-Type-Options": "nosniff",  # 禁止浏览器自动猜测MIME类型
            "X-Frame-Options": "DENY",            # 禁止被iframe嵌入
            "X-XSS-Protection": "1; mode=block", # 开启XSS防护
            "Referrer-Policy": "strict-origin-when-cross-origin", # 限制Referrer
        })
        return response

app.add_middleware(SecurityHeadersMiddleware)

2. 统一错误处理中间件

捕获所有未处理的异常，返回统一的JSON错误格式，避免暴露内部信息。

from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
from starlette.responses import JSONResponse
import traceback
import os

DEBUG = os.getenv("ENVIRONMENT") != "production"

class UnifiedErrorHandler(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        try:
            return await call_next(request)
        # 处理FastAPI验证错误
        except RequestValidationError as e:
            return JSONResponse(
                status_code=422,
                content={"code": 422, "msg": "参数验证失败", "detail": e.errors()}
            )
        # 处理HTTP异常
        except StarletteHTTPException as e:
            return JSONResponse(
                status_code=e.status_code,
                content={"code": e.status_code, "msg": e.detail}
            )
        # 处理未捕获的内部错误
        except Exception as e:
            if DEBUG:
                return JSONResponse(
                    status_code=500,
                    content={"code": 500, "msg": "内部错误", "traceback": traceback.format_exc()}
                )
            else:
                return JSONResponse(
                    status_code=500,
                    content={"code": 500, "msg": "服务器开小差了，请稍后再试"}
                )

app.add_middleware(UnifiedErrorHandler)

五、生产环境最佳实践

1. 控制中间件顺序

请求时按注册顺序执行，响应时按反向顺序执行，要注意：

  • CORS中间件要放在最前面（处理预检请求）
  • 错误处理中间件要放在最后面（捕获所有上层中间件/路由的错误）
# ✅ 正确的注册顺序
app.add_middleware(SecurityHeadersMiddleware)
app.add_middleware(CORSMiddleware, **cors_config)
app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(PerformanceMonitoringMiddleware)  # 假设已定义
app.add_middleware(UnifiedErrorHandler)  # 必须放在最后

2. 避免中间件性能陷阱

  • 不要在中间件里做耗时操作：比如数据库查询、外部API调用（除非是必要的全局检查）
  • 快速跳过无关路径：健康检查、文档路径等不需要处理
  • 异步处理所有IO操作：确保中间件里的代码是异步的，避免阻塞整个事件循环

六、总结

FastAPI中间件是实现全局横切关注点的核心工具，掌握它可以显著提升应用的安全性、性能和可维护性：

  1. 优先使用官方/内置中间件（CORS、GZip），避免重复造轮子
  2. 轻量功能用装饰器中间件，复杂功能用BaseHTTPMiddleware
  3. 严格控制中间件顺序，合理设置跳过路径
  4. 生产环境要充分测试，避免性能陷阱

🔗 扩展阅读