FastAPI: 中间件 — 请求/响应的拦截与增强
最后更新：2026-08-26
中间件就像机场的安检通道——每个旅客（请求）必须依次通过海关、安检、登机口，任何一站可以放行也可以拦截，而回程（响应）按相反顺序经过。
1. 你将学到
- Starlette 中间件工作原理：ASGI callable 包装链
- 内置中间件：
CORSMiddleware配置与安全策略 - 自定义中间件：请求计时、请求 ID 注入、响应头注入
- 中间件执行顺序：注册顺序与实际执行顺序的关系
- Alice 场景：为 PriceTracker 添加请求日志中间件与 API 限流中间件
2. Alice 的真实故事
(1) 痛点：API 被恶意刷量且无法追踪
Alice 的 PriceTracker API 上线后，发现某个 IP 每分钟发送 5000 次请求，导致数据库负载飙升。更糟糕的是，Bob 前端从 localhost:3000 调用 API 时被浏览器 CORS 策略拦截，所有请求失败。Alice 需要同时解决跨域访问和请求限流两个问题，但 Flask 没有统一的中间件机制。
(2) FastAPI 中间件的解法
FastAPI/Starlette 提供洋葱模型中间件，一层解决一个问题：CORS 中间件处理跨域，限流中间件控制频率，日志中间件记录请求——互不干扰，注册即生效。
PYTHON
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware
app = FastAPI()
app.add_middleware(CORSMiddleware, allow_origins=["http://localhost:3000"])
(3) 收益
Bob 前端跨域问题 5 行代码解决，限流中间件将恶意 IP 的请求从每分钟 5000 次降到 1000 次，日志中间件让每个请求都有唯一 ID 方便追踪。
3. 中间件洋葱模型
(1) 洋葱模型原理
flowchart TD
Request[Client Request] --> CORS[CORS Middleware]
CORS --> Logging[Logging Middleware]
Logging --> RateLimit[Rate Limit Middleware]
RateLimit --> App[FastAPI Application]
App --> RateLimit2[Rate Limit Response]
RateLimit2 --> Logging2[Logging Response]
Logging2 --> CORS2[CORS Response]
CORS2 --> Response[Client Response]
RateLimit -.->|429 Too Many Requests| Reject[Short-circuit Rejection]
Reject --> Logging2
| 特性 | 说明 |
|---|---|
| 请求方向 | 从外到内（最先注册的最外层） |
| 响应方向 | 从内到外（与请求相反） |
| 短路能力 | 任何中间件可提前返回响应，不进入下一层 |
| 注册顺序 | 最后注册的中间件最先处理请求 |
▶ 示例：中间件注册顺序与执行顺序
PYTHON
from fastapi import FastAPI, Request
import time
app = FastAPI()
# Middleware registered FIRST = outermost layer (executes first on request)
@app.middleware("http")
async def outer_middleware(request: Request, call_next):
print("Outer: before request")
response = await call_next(request)
print("Outer: after response")
return response
# Middleware registered LAST = innermost layer (executes last on request)
@app.middleware("http")
async def inner_middleware(request: Request, call_next):
print("Inner: before request")
response = await call_next(request)
print("Inner: after response")
return response
@app.get("/test")
async def test():
print("Handler: processing")
return {"ok": True}
输出：
TEXT
📖 仅展示
Outer: before request
Inner: before request
Handler: processing
Inner: after response
Outer: after response
4. CORS 中间件
(1) 跨域资源共享配置
CORS（Cross-Origin Resource Sharing）是浏览器安全策略，限制不同源的网页访问 API。Bob 的前端（localhost:3000）访问 Alice 的 API（localhost:8000）属于跨域。
▶ 示例：开发环境 CORS 配置
PYTHON
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000"], # Bob's frontend
allow_credentials=True,
allow_methods=["*"], # All HTTP methods
allow_headers=["*"], # All headers
)
输出：
TEXT
📖 仅展示
# 执行成功
▶ 示例：生产环境 CORS 配置
PYTHON
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware
app = FastAPI()
ALLOWED_ORIGINS = [
"https://pricetracker.example.com",
"https://admin.pricetracker.example.com",
]
app.add_middleware(
CORSMiddleware,
allow_origins=ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["Authorization", "Content-Type"],
)
输出：
TEXT
📖 仅展示
# 执行成功
| CORS 参数 | 开发环境 | 生产环境 |
|---|---|---|
allow_origins |
["*"] 或 localhost |
具体域名列表 |
allow_credentials |
True | True（如需 Cookie） |
allow_methods |
["*"] |
仅需要的方法 |
allow_headers |
["*"] |
仅需要的头 |
max_age |
默认 | 3600（缓存预检） |
5. 自定义中间件
(1) 请求计时中间件
▶ 示例：记录每个请求的处理时间
PYTHON
from fastapi import FastAPI, Request
import time
app = FastAPI()
@app.middleware("http")
async def timing_middleware(request: Request, call_next):
start_time = time.perf_counter()
response = await call_next(request)
process_time = time.perf_counter() - start_time
response.headers["X-Process-Time"] = f"{process_time:.4f}s"
return response
@app.get("/products")
async def list_products():
return [{"id": 1, "name": "Widget"}]
输出：
TEXT
📖 仅展示
# 函数定义成功
▶ 示例：请求 ID 注入中间件
PYTHON
import uuid
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def request_id_middleware(request: Request, call_next):
request_id = str(uuid.uuid4())
request.state.request_id = request_id # Attach to request state
response = await call_next(request)
response.headers["X-Request-ID"] = request_id
return response
@app.get("/health")
async def health_check(request: Request):
return {
"status": "healthy",
"request_id": request.state.request_id,
}
输出：
TEXT
📖 仅展示
# 函数定义成功
▶ 示例：API 限流中间件（简易版）
PYTHON
from fastapi import FastAPI, Request, HTTPException
from collections import defaultdict
import time
app = FastAPI()
# Simple in-memory rate limiter
rate_limits: dict[str, list[float]] = defaultdict(list)
RATE_LIMIT = 1000 # requests per minute
WINDOW = 60 # seconds
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
client_ip = request.client.host if request.client else "unknown"
now = time.time()
# Clean old entries
rate_limits[client_ip] = [
t for t in rate_limits[client_ip] if now - t < WINDOW
]
# Check limit
if len(rate_limits[client_ip]) >= RATE_LIMIT:
raise HTTPException(
status_code=429,
detail=f"Rate limit exceeded: {RATE_LIMIT} requests per {WINDOW}s",
)
rate_limits[client_ip].append(now)
response = await call_next(request)
response.headers["X-RateLimit-Limit"] = str(RATE_LIMIT)
response.headers["X-RateLimit-Remaining"] = str(
RATE_LIMIT - len(rate_limits[client_ip])
)
return response
输出：
TEXT
📖 仅展示
# 函数定义成功
6. 类式中间件与 ASGI callable
(1) 类式中间件写法
对于更复杂的中间件逻辑，推荐用类式写法，直接操作 ASGI scope/receive/send。
▶ 示例：类式请求日志中间件
PYTHON
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
import logging
import time
logger = logging.getLogger("pricetracker")
class LoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next) -> Response:
start = time.perf_counter()
logger.info(f"Request: {request.method} {request.url.path}")
response = await call_next(request)
duration = time.perf_counter() - start
logger.info(
f"Response: {request.method} {request.url.path} "
f"status={response.status_code} duration={duration:.4f}s"
)
return response
# Register class-based middleware
app = FastAPI()
app.add_middleware(LoggingMiddleware)
输出：
TEXT
📖 仅展示
# 函数定义成功
(2) 中间件类型对比
| 写法 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
@app.middleware("http") |
简单拦截 | 代码少 | 仅处理 HTTP |
BaseHTTPMiddleware |
中等复杂度 | 可配置化 | 仅处理 HTTP |
| 纯 ASGI 类 | 完全控制 | WebSocket 也处理 | 代码复杂 |
❓ 常见问题
Q 中间件注册顺序为什么重要？
A 最后注册的中间件最先处理请求（最内层）。CORS 应最晚注册（最外层），确保所有响应都带 CORS 头。限流应较早注册（较内层），被拒绝的请求仍经过 CORS。
Q @app.middleware 和 add_middleware 能混用吗？
A 可以，但
@app.middleware 注册的中间件在 add_middleware 之前（更外层）。建议统一用一种写法。Q 生产限流用什么方案？
A 简易内存限流仅适合单进程。生产环境用 Redis + slowapi 或自定义 Redis 限流中间件，支持分布式和多进程。
Q CORS 的 allow_origins 能用 * 吗？
A 开发环境可以，生产环境必须指定具体域名。
allow_credentials=True 时不能用 *，浏览器会拒绝。Q 中间件能修改请求体吗？
A 可以但需要先读取 body 再重新构造。BaseHTTPMiddleware 有已知问题（读取 body 后 stream 消耗），复杂场景建议用纯 ASGI 中间件。
Q 如何禁用某个中间件？
A 没有内置开关。建议用环境变量控制：
if settings.ENABLE_RATE_LIMIT: app.add_middleware(RateLimitMiddleware)。📖 小节
- 中间件遵循洋葱模型：请求从外到内，响应从内到外，任何层可短路返回
CORSMiddleware解决跨域问题，生产环境必须指定具体域名- 自定义中间件可实现请求计时、请求 ID 注入、限流等横切关注点
- 中间件注册顺序决定执行顺序：最后注册的最先处理请求（最内层）
- 类式中间件（
BaseHTTPMiddleware）适合复杂逻辑，纯 ASGI 类可处理 WebSocket
📝 作业
- 基础题（难度⭐）：为 PriceTracker 添加 CORS 中间件，允许
http://localhost:3000跨域访问，在浏览器中验证 Bob 前端可以调用 API。提示：app.add_middleware(CORSMiddleware, ...) - 进阶题（难度⭐⭐）：实现请求计时中间件，在响应头中添加
X-Process-Time，并用 Swagger UI 观察响应头。提示：@app.middleware("http")+time.perf_counter() - 挑战题（难度⭐⭐⭐）：实现基于 IP 的限流中间件（每分钟 1000 次），超限时返回 429 状态码和
Retry-After响应头，同时在响应头中显示剩余配额。提示：defaultdict(list)+ 时间窗口清理
---|
