FastAPI 中间件
FastAPI 中间件学习笔记
一、中间件概念
中间件是一个在请求到达路由处理函数之前和响应返回客户端之前执行的函数，类似于一个"拦截器"：
客户端请求 → 中间件1 → 中间件2 → ... → 路由处理函数 ↓ 客户端响应 ← 中间件1 ← 中间件2 ← ... ← 路由处理函数返回核心作用：在请求/响应的流转过程中插入通用逻辑，如日志、认证、CORS、计时等。
二、基本用法 —@app.middleware
1. 最简示例
fromfastapiimportFastAPI,Request app=FastAPI()@app.middleware("http")asyncdefsimple_middleware(request:Request,call_next):# ---- 请求阶段（before route handler）----print(f"请求进入:{request.method}{request.url}")# ---- 调用下一个中间件或路由处理函数 ----response=awaitcall_next(request)# ---- 响应阶段（after route handler）----print(f"响应状态:{response.status_code}")returnresponse2. 执行流程
@app.middleware("http") async def my_middleware(request, call_next): # ① 请求前逻辑 response = await call_next(request) # ② 交给下一层 # ③ 响应后逻辑 return response # ④ 返回响应| 阶段 | 位置 | 典型用途 |
|---|---|---|
| ① 请求前 | call_next之前 | 认证、限流、注入上下文 |
| ② 传递 | call_next(request) | 调用下一个中间件/路由 |
| ③ 响应后 | call_next之后 | 添加响应头、日志记录 |
| ④ 返回 | return response | 将处理后的响应返回 |
三、实用示例
1. 请求计时中间件
importtime@app.middleware("http")asyncdefadd_process_time_header(request:Request,call_next):start_time=time.perf_counter()response=awaitcall_next(request)process_time=time.perf_counter()-start_time response.headers["X-Process-Time"]=str(process_time)returnresponse2. 认证中间件
fromfastapiimportRequest,HTTPExceptionfromstarlette.responsesimportJSONResponse@app.middleware("http")asyncdefauth_middleware(request:Request,call_next):# 白名单路径跳过认证ifrequest.url.pathin("/docs","/openapi.json","/login"):returnawaitcall_next(request)token=request.headers.get("Authorization")ifnottokenornotverify_token(token):returnJSONResponse(status_code=401,content={"detail":"Not authenticated"},)# 认证通过，继续处理response=awaitcall_next(request)returnresponsedefverify_token(token:str)->bool:# 实际项目中验证 JWT / Sessionreturntoken=="Bearer valid-token"3. 请求日志中间件
importloggingimporttime logger=logging.getLogger("api.access")@app.middleware("http")asyncdeflogging_middleware(request:Request,call_next):start=time.perf_counter()response=awaitcall_next(request)duration=time.perf_counter()-start logger.info(f"{request.method}{request.url.path}"f"status={response.status_code}"f"duration={duration:.3f}s "f"client={request.client.hostifrequest.clientelse'unknown'}")returnresponse4. 请求 ID 追踪中间件
importuuid@app.middleware("http")asyncdefrequest_id_middleware(request:Request,call_next):request_id=request.headers.get("X-Request-ID")orstr(uuid.uuid4())response=awaitcall_next(request)response.headers["X-Request-ID"]=request_idreturnresponse四、中间件执行顺序
中间件按注册顺序执行请求阶段，按逆序执行响应阶段（洋葱模型）：
@app.middleware("http")asyncdefmiddleware_a(request,call_next):print("A 请求前")response=awaitcall_next(request)print("A 响应后")returnresponse@app.middleware("http")asyncdefmiddleware_b(request,call_next):print("B 请求前")response=awaitcall_next(request)print("B 响应后")returnresponse输出顺序：
A 请求前 → B 请求前 → [路由处理] → B 响应后 → A 响应后┌─── middleware_a ───┐ │ ┌── middleware_b ┐ │ │ │ [route handler] │ │ │ └─────────────────┘ │ └──────────────────────┘注意：
@app.middleware装饰器注册的中间件，后注册的先执行请求阶段（栈结构）。实际执行顺序与代码书写顺序相反，需留意。
五、call_next详解
1. 基本行为
response=awaitcall_next(request)- 调用
call_next将请求传递给下一层，返回Response对象。 - 必须调用，否则请求不会到达路由处理函数。
2. 修改请求体
call_next接收的是Request对象，如需修改请求体，需构造新的Request：
importjson@app.middleware("http")asyncdefmodify_body_middleware(request:Request,call_next):# 读取原始请求体body=awaitrequest.body()# 修改后构造新请求modified_body=json.dumps({"injected":True,**json.loads(body)}).encode()new_request=Request(request.scope,receive=lambda:receive_with_body(modified_body),)response=awaitcall_next(new_request)returnresponse修改请求体场景较少见，通常建议用依赖注入（
Depends）替代。
3. 提前返回响应
在call_next之前直接返回响应，请求不会到达路由：
@app.middleware("http")asyncdefrate_limit_middleware(request:Request,call_next):ifis_rate_limited(request):returnJSONResponse(status_code=429,content={"detail":"Too many requests"})returnawaitcall_next(request)六、内置中间件 — CORS
FastAPI（Starlette）提供了多个内置中间件，最常用的是 CORS：
fromfastapi.middleware.corsimportCORSMiddleware app.add_middleware(CORSMiddleware,allow_origins=["https://example.com","http://localhost:3000"],allow_credentials=True,allow_methods=["*"],allow_headers=["*"],)| 参数 | 说明 |
|---|---|
allow_origins | 允许的源列表，["*"]表示全部 |
allow_methods | 允许的 HTTP 方法 |
allow_headers | 允许的请求头 |
allow_credentials | 是否允许携带 Cookie |
max_age | 预检请求缓存时间（秒） |
七、其他内置中间件
| 中间件 | 用途 | 示例 |
|---|---|---|
CORSMiddleware | 跨域资源共享 | 见上方 |
HTTPSRedirectMiddleware | 强制 HTTPS | app.add_middleware(HTTPSRedirectMiddleware) |
TrustedHostMiddleware | 主机白名单 | 防止 Host 头攻击 |
GZipMiddleware | 响应压缩 | 减少传输体积 |
SessionMiddleware | Cookie-based Session | 基于 Cookie 的会话管理 |
fromstarlette.middleware.httpsredirectimportHTTPSRedirectMiddlewarefromstarlette.middleware.trustedhostimportTrustedHostMiddlewarefromstarlette.middleware.gzipimportGZipMiddleware app.add_middleware(GZipMiddleware,minimum_size=1000)app.add_middleware(TrustedHostMiddleware,allowed_hosts=["example.com","*.example.com"])app.add_middleware(HTTPSRedirectMiddleware)
add_middleware注册顺序：与@app.middleware相反，add_middleware先添加的在外层。建议将 CORS 放在最外层。
八、中间件 vs 依赖注入
| 对比项 | 中间件 | 依赖注入（Depends） |
|---|---|---|
| 作用范围 | 全局，所有请求 | 可按路由/路径选择性应用 |
| 执行时机 | 路由匹配之前 | 路由匹配之后 |
| 能否修改响应 | 可以 | 不方便 |
| 能否提前返回 | 可以 | 可以（抛 HTTPException） |
| 典型场景 | 日志、CORS、压缩、全局认证 | 参数校验、数据库会话、权限检查 |
| 可读性 | 集中管理 | 声明式，与路由绑定 |
选择建议：
- 全局通用逻辑→ 中间件（如日志、CORS、压缩）
- 路由级逻辑→ 依赖注入（如数据库连接、特定权限校验）
九、纯 ASGI 中间件
除了 FastAPI 的@app.middleware，还可以编写纯 ASGI 中间件，兼容任何 ASGI 框架：
classCustomASGIMiddleware:def__init__(self,app):self.app=appasyncdef__call__(self,scope,receive,send):ifscope["type"]=="http":# 请求前逻辑print("ASGI middleware: before request")awaitself.app(scope,receive,send)ifscope["type"]=="http":# 请求后逻辑print("ASGI middleware: after request")# 注册app.add_middleware(CustomASGIMiddleware)纯 ASGI 中间件更底层、更灵活，但编写复杂。大多数场景用
@app.middleware即可。
十、注意事项
- 中间件中不要抛 HTTPException：中间件中
raise HTTPException不会被异常处理器捕获（因为异常处理器在路由层之后）。应直接返回JSONResponse。 - 避免在中间件中读取请求体：
await request.body()会消费请求体流，后续路由无法再次读取。如必须读取，需重新构造Request。 - 中间件顺序很重要：CORS 通常放最外层，认证放内层，日志放最外层。
- 性能考量：每个请求都会经过所有中间件，中间件中的阻塞操作会拖慢全局响应。
- 测试时注意：中间件会影响测试结果，
TestClient会完整执行中间件链。