Denny的博客

Denny的博客

0%

FastAPI自学教程（62） - 中间件机制详解

发表于 分类于

1.基础用法

1
2
3
4
5
6
7
8
9
10
11
12
from fastapi import FastAPI, Request
import time

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response
  • 中间件通过@app.middleware("http")装饰器声明，或通过add_middleware()方法添加
  • 每个中间件接收request对象和call_next回调函数，可在请求处理前后插入逻辑
  • 核心处理流程： 
    graph TD
  A[请求到达] --> B[中间件预处理]
  B --> C{是否继续?}
  C -->|是| D[调用后续中间件/路由]
  C -->|否| E[直接返回响应]
  D --> F[中间件后处理]
  F --> G[返回客户端]

2.进阶配置

2.1 自定义中间件类

1
2
3
4
5
6
7
8
from starlette.middleware.base import BaseHTTPMiddleware

class CustomMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 请求前处理
        response = await call_next(request)
        # 响应后处理
        return response
  • 继承BaseHTTPMiddleware类，实现dispatch方法
  • 支持获取请求体：request_body = await request.body()

2.2 执行顺序控制

1
2
app.add_middleware(MiddlewareA)  # 第二执行
app.add_middleware(MiddlewareB)  # 第一执行
  • 中间件按添加顺序逆序执行（后添加的先执行）
  • 建议顺序：安全类中间件 → 业务中间件 → 性能监控中间件

3.内置中间件示例

3.1 跨域处理

1
2
3
4
5
6
7
8
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"]
)
  • 配置参数支持正则匹配、指定域名等

3.2 会话管理

1
2
3
4
5
6
7
from starlette.middleware.sessions import SessionMiddleware

app.add_middleware(
    SessionMiddleware,
    secret_key="your-secret-key",
    session_cookie="session"
)
  • 支持设置会话过期时间、SameSite策略等

4.生产环境建议

  1. 安全加固：

    • 添加X-Content-Type-Options和Content-Security-Policy响应头
    • 敏感操作中间件应验证request.client.host来源

  2. 性能优化：

    • 使用GZipMiddleware压缩响应体（设置minimum_size=1024过滤小文件）
    • 流式处理大文件避免内存溢出

  3. 错误监控： 

    1
    2
    3
    4
    5
    6
    7
    @app.middleware("http")
    async def catch_exceptions(request: Request, call_next):
        try:
            return await call_next(request)
        except Exception as exc:
            logger.error(f"Error processing request: {exc}")
            return JSONResponse({"detail": "Internal error"}, 500)
    • 全局捕获未处理异常

测试命令示例： 

1
2
# 测试跨域中间件
curl -H "Origin: http://localhost:3000" -v http://localhost:8000/