06-中间件

Python 3.11+

本章讲解 FastAPI 中间件和跨域处理。

第一部分：内置中间件

1.1 实际场景

前端应用运行在 localhost:3000，后端 API 运行在 localhost:8000，浏览器会阻止跨域请求。

问题：如何解决跨域问题？

1.2 CORS 中间件

python

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

app: FastAPI = FastAPI()

# 添加 CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

1.3 GZip 中间件

python

from fastapi.middleware.gzip import GZipMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)

第二部分：自定义中间件

2.1 实际场景

需要在每个请求中添加处理时间响应头，记录请求日志。

问题：如何创建自定义中间件？

2.2 基础中间件

python

from fastapi import FastAPI, Request
import time

app: FastAPI = FastAPI()


@app.middleware("http")
async def add_process_time(request: Request, call_next):
    start_time: float = time.time()
    response = await call_next(request)
    process_time: float = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

2.3 请求日志中间件

python

@app.middleware("http")
async def log_requests(request: Request, call_next):
    print(f"Request: {request.method} {request.url}")
    response = await call_next(request)
    print(f"Response: {response.status_code}")
    return response

第三部分：中间件进阶

3.1 实际场景

所有 API 请求需要验证 Token，除了登录和文档页面。

问题：如何在中间件中实现认证逻辑？

3.2 认证中间件

python

from fastapi import FastAPI, Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 公开路径
        public_paths: list[str] = ["/docs", "/openapi.json", "/token"]
        
        if request.url.path in public_paths:
            return await call_next(request)
        
        # 检查认证
        auth_header: str | None = request.headers.get("Authorization")
        if not auth_header or not auth_header.startswith("Bearer "):
            raise HTTPException(status_code=401, detail="Not authenticated")
        
        return await call_next(request)

app.add_middleware(AuthMiddleware)

第四部分：完整示例

python

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
import time

app: FastAPI = FastAPI()

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# GZip
app.add_middleware(GZipMiddleware, minimum_size=500)

# 自定义中间件
@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    start: float = time.time()
    response = await call_next(request)
    process_time: float = time.time() - start
    response.headers["X-Process-Time"] = f"{process_time:.4f}"
    return response


@app.middleware("http")
async def security_headers(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    return response


@app.get("/")
async def root() -> dict[str, str]:
    return {"message": "Hello"}


@app.get("/slow")
async def slow() -> dict[str, str]:
    time.sleep(2)
    return {"message": "Done"}

第五部分：L3 专家层

5.1 ASGI 中间件的洋葱模型（Stacking Middleware）

FastAPI 中间件基于 ASGI 规范，采用洋葱模型：请求从外向内逐层穿过中间件，响应从内向外逐层返回。

         ┌──────────────────────────────────────┐
         │        Middleware A (CORS)           │
         │  ┌────────────────────────────────┐  │
         │  │    Middleware B (GZip)         │  │
         │  │  ┌──────────────────────────┐  │  │
         │  │  │   Middleware C (Timing)  │  │  │
         │  │  │  ┌────────────────────┐  │  │  │
    请求 →│→│→│→│→→│     Router/App     │→→│←│←│←│ 响应
         │  │  │  └────────────────────┘  │  │  │
         │  │  │   Middleware C (Timing)  │  │  │
         │  │  └──────────────────────────┘  │  │
         │  │    Middleware B (GZip)         │  │
         │  └────────────────────────────────┘  │
         │        Middleware A (CORS)           │
         └──────────────────────────────────────┘

    请求方向：外层 → 内层 (A → B → C → App)
    响应方向：内层 → 外层 (App → C → B → A)

python

# 洋葱模型本质：每个中间件都是 awaitable 的嵌套调用
# 伪代码表示
async def middleware_a(scope, receive, send):
    # 请求前处理（外层）
    async def wrapped_send(message):
        # 响应后处理（外层）
        await send(message)
    await middleware_b(scope, receive, wrapped_send)

async def middleware_b(scope, receive, send):
    async def wrapped_send(message):
        await send(message)
    await middleware_c(scope, receive, wrapped_send)

5.2 Starlette 中间件的执行顺序

FastAPI 继承自 Starlette，中间件的执行顺序与注册顺序相反（后注册的更靠近 App，先处理请求）：

python

app.add_middleware(A)  # 最外层（最后处理请求）
app.add_middleware(B)  # 中间层
app.add_middleware(C)  # 最内层（最先处理请求，最先接触响应）

# 请求流：Client → C → B → A → App → A → B → C → Client
#                          ↑ 最先到达       ↑ 最后离开

中间件	注册顺序	请求阶段位置	响应阶段位置
CORSMiddleware	第一个	最外层（最后执行）	最外层（最后修改）
GZipMiddleware	第二个	中间层	中间层
自定义中间件	最后一个	最内层（最先执行）	最内层（最先修改）

关键规则：

CORS 必须最先注册（最外层），确保 OPTIONS 预检请求被优先处理
GZip 应靠近 App（内层），在响应生成后再压缩
认证/日志中间件应放在 CORS 之后

@app.middleware("http") 内部通过 Middleware 包装为 BaseHTTPMiddleware：

python

# fastapi/applications.py 简化版
def middleware(self, type: str = "http"):
    def decorator(func):
        self.add_middleware(Middleware, base_http_middleware_class=..., dispatch=func)
        return func
    return decorator

5.3 CORS 预检请求（Preflight）处理

浏览器对非简单请求（方法非 GET/POST/HEAD，或包含自定义 Header）会先发送 OPTIONS 预检请求：

Client                              Server
  │── OPTIONS /api/users ──────────→│
  │   Origin: http://localhost:3000 │
  │   Access-Control-Request-Method: DELETE
  │   Access-Control-Request-Headers: Authorization
  │                                 │
  │←─ 200 OK ──────────────────────│
  │   Access-Control-Allow-Origin: http://localhost:3000
  │   Access-Control-Allow-Methods: DELETE
  │   Access-Control-Allow-Headers: Authorization
  │   Access-Control-Max-Age: 600
  │                                 │
  │── DELETE /api/users ───────────→│  ← 实际请求
  │   Authorization: Bearer <token> │
  │←─ 204 No Content ──────────────│

python

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from starlette.responses import PlainTextResponse

app: FastAPI = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type"],
    max_age=600,  # 预检结果缓存 600 秒
)

# CORSMiddleware 内部处理 OPTIONS 的逻辑（简化）
# class CORSMiddleware:
#     async def __call__(self, scope, receive, send):
#         if method == "OPTIONS" and "origin" in headers:
#             # 直接返回 CORS headers，不经过后续中间件和路由
#             response = PlainTextResponse("", headers=cors_headers)
#             await response(scope, receive, send)
#             return

触发预检的条件	不触发预检的条件
自定义 Header（如 `Authorization`）	`Accept`, `Accept-Language`, `Content-Language`
`Content-Type: application/json`	`Content-Type: application/x-www-form-urlencoded`
方法为 `PUT`、`DELETE`、`PATCH`	方法为 `GET`、`HEAD`、`POST`
携带凭证的跨域请求	同源请求

5.4 性能考量

操作	耗时级别	说明
CORS 预检检查	~0.001ms	纯字符串比对
洋葱模型层间传递	~0.01ms/层	每次 `call_next` 增加一层 await
GZip 压缩（1MB）	~5-10ms	CPU 密集型，`minimum_size` 应设合理值
中间件数量（10 层）	~0.1ms 总开销	通常可忽略，但层数多时需关注

5.5 设计动机

设计选择	动机
`add_middleware` 反向注册顺序	后注册的中间件更接近 App，符合"先注册先处理全局事务"的直觉
`BaseHTTPMiddleware` 封装	简化 ASGI 原生接口，提供 `request`/`response` 高级对象
`@app.middleware("http")` 装饰器	零配置快速添加中间件，内部自动包装为 `BaseHTTPMiddleware`
CORS 拦截 OPTIONS	预检请求不需要经过业务路由，提前返回减少不必要的处理开销

5.6 知识关联

中间件
├── ASGI 规范
│   ├── scope: dict（请求信息）
│   ├── receive: Awaitable（接收消息）
│   └── send: Callable（发送响应）
│
├── 洋葱模型（Stacking）
│   ├── 请求：外层 → 内层 → App
│   └── 响应：App → 内层 → 外层
│
├── Starlette 中间件类型
│   ├── CORSMiddleware：跨域资源共享
│   ├── GZipMiddleware：响应压缩
│   ├── TrustedHostMiddleware：Host 验证
│   └── BaseHTTPMiddleware：自定义基类
│
├── CORS 预检（Preflight）
│   ├── 触发条件：非简单方法、自定义 Header、JSON Content-Type
│   ├── 缓存：max_age 控制 OPTIONS 结果缓存时间
│   └── 凭证：allow_credentials 控制 Cookie/Authorization 传递
│
└── 中间件注册顺序
    ├── 最先注册：最外层（CORS 优先）
    ├── 最后注册：最内层（靠近 App）
    └── @app.middleware：最内层（最后注册）

总结

知识点	说明
CORSMiddleware	跨域请求
GZipMiddleware	压缩
@app.middleware	自定义中间件
请求/响应处理	中间件钩子

06-中间件 ​

第一部分：内置中间件 ​

1.1 实际场景 ​

1.2 CORS 中间件 ​

1.3 GZip 中间件 ​

第二部分：自定义中间件 ​

2.1 实际场景 ​

2.2 基础中间件 ​

2.3 请求日志中间件 ​

第三部分：中间件进阶 ​

3.1 实际场景 ​

3.2 认证中间件 ​

第四部分：完整示例 ​

第五部分：L3 专家层 ​

5.1 ASGI 中间件的洋葱模型（Stacking Middleware） ​

5.2 Starlette 中间件的执行顺序 ​

5.3 CORS 预检请求（Preflight）处理 ​

5.4 性能考量 ​

5.5 设计动机 ​

5.6 知识关联 ​

总结 ​

06-中间件

第一部分：内置中间件

1.1 实际场景

1.2 CORS 中间件

1.3 GZip 中间件

第二部分：自定义中间件

2.1 实际场景

2.2 基础中间件

2.3 请求日志中间件

第三部分：中间件进阶

3.1 实际场景

3.2 认证中间件

第四部分：完整示例

第五部分：L3 专家层

5.1 ASGI 中间件的洋葱模型（Stacking Middleware）

5.2 Starlette 中间件的执行顺序

5.3 CORS 预检请求（Preflight）处理

5.4 性能考量

5.5 设计动机

5.6 知识关联

总结