25. FastAPI 中间件

什么是中间件

中间件是包在应用外层的一段处理程序，所有 HTTP 请求进出时都要经过它。它可以在路由执行前读取请求，在路由执行后修改响应。

可以把中间件想象成大楼的收发室：无论来访者要去哪个房间，进门都要先经过这里登记，出门时也在这里交还访客牌。收发室不关心访客去办什么业务，但能给整栋楼统一加上登记、安检这类公共动作。

一次中间件调用可以拆成两个阶段，call_next 是分界线：

阶段	能做什么
call_next 之前	读取请求头、解析或记录信息、判断是否放行、直接返回拒绝响应
call_next 之后	拿到响应、统一加响应头、记录状态码和耗时、替换或包装响应

call_next 把请求交给内层的应用（可能是下一层中间件或路由），并拿到最终响应对象。

中间件的使用场景和作用

中间件适合放"与具体业务无关、但很多请求都需要"的横切关注点。典型场景：

• 访问日志、请求编号、耗时统计。
• 限流、压缩、安全响应头、跨域（见第 26 篇）。
• 统一响应格式、错误兜底。

不适合的场景：

• 某个业务自身的规则（库存扣减、下单、价格计算）。中间件无法自然地拿到"这是哪个商品的库存"，硬塞进去会让所有请求都背上不相干的逻辑。
• 需要路径参数或请求体字段做判断。中间件拿的是原始 Request，路由参数要等路由匹配后才解析，硬做等于重写一套解析逻辑。

一个可操作的判断：如果去掉"HTTP 请求"这个概念后这段逻辑仍能成立（比如数据库连接、密码哈希），它多半属于依赖或业务层；如果它天然就是对请求和响应做包装（比如加响应头、记录耗时、统一返回结构），才更像中间件。

中间件的执行顺序

连续注册多层中间件时，后注册的更靠外：请求先经过最后注册的那层，响应按相反方向返回。

用一个最小案例验证：

from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def a(request: Request, call_next):
    print("A：进入")
    response = await call_next(request)
    print("A：退出")
    return response

@app.middleware("http")
async def b(request: Request, call_next):
    print("B：进入")
    response = await call_next(request)
    print("B：退出")
    return response

@app.get("/demo")
def demo():
    print("路由：处理业务")
    return {"ok": True}

请求 /demo 时，服务端控制台按顺序打印：

B：进入
A：进入
路由：处理业务
A：退出
B：退出

a 先注册、b 后注册，后注册的 b 反而包在最外层，所以请求先经过 b。执行顺序像套在一起的盒子：

这里用 print 验证顺序，而不是把结果写进响应体，因为退出阶段更适合处理响应头、日志、耗时这类动作，硬改响应体容易和路由返回冲突。

call_next 将请求交给内层应用并取得响应对象。对流式响应而言，此时未必已经发送完整正文，所以耗时不能当作"客户端下载完全部流"的总耗时。

FastAPI 里的三个级别

先厘清一个关键点：FastAPI 中真正叫"中间件"的只有全局这一种，通过 add_middleware 或 @app.middleware("http") 挂载。APIRouter 本身没有 middleware 参数，所以"模块级"和"路由级"通常用依赖注入 Depends 来实现——它同样能在请求进入路由前"插一手"。

因此三个级别实际对应两套机制：

级别	机制	挂载位置	作用范围
全局级	中间件	app.add_middleware / @app.middleware	所有请求
模块级	依赖 Depends	APIRouter(dependencies=[...])	该路由器的全部接口
路由级	依赖 Depends	路径操作 dependencies=[...] 或函数参数	单个接口

下面逐级说明，重点讲依赖实现（依赖注入从第 31 篇开始有系统介绍，这里先看用法）。

全局级中间件

全局中间件对所有请求生效，适合访问日志、请求编号、耗时统计这类横切动作。

from time import perf_counter
from uuid import uuid4
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def trace_request(request: Request, call_next):
    request_id = uuid4().hex
    request.state.request_id = request_id
    started = perf_counter()
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Process-Time"] = f"{perf_counter() - started:.6f}"
    return response

@app.get("/hello")
def hello(request: Request):
    return {"message": "你好", "request_id": request.state.request_id}

保存为 main.py，运行 uv run uvicorn main:app --reload。执行 curl.exe -i http://127.0.0.1:8000/hello，响应头与正文中的编号应一致。

模块级依赖（Router 级）

模块级依赖挂在整个 APIRouter 上，对该路由器下所有接口生效，适合"整个模块都要校验"的场景，例如一个租户模块的所有接口都要带租户头。

from typing import Annotated
from fastapi import APIRouter, Depends, Header, HTTPException

async def require_tenant(x_tenant: Annotated[str, Header()] = ""):
    if not x_tenant:
        raise HTTPException(400, "缺少 X-Tenant 请求头")
    return {"tenant": x_tenant}

users = APIRouter(prefix="/users", tags=["用户"], dependencies=[Depends(require_tenant)])

@users.get("/{user_id}")
def detail(user_id: int):
    return {"id": user_id, "name": "示例用户"}

@users.post("")
def create():
    return {"created": True}

users 下的 /users/1、POST /users 都会先执行 require_tenant。只有这个模块需要租户校验时，用模块级依赖比全局中间件更精准，不会连累其他模块。

路由级依赖

路由级依赖只作用于单个接口，适合"某个接口需要额外检查"的场景，例如只有删除接口需要管理员权限。

from fastapi import APIRouter, Depends, HTTPException

def require_admin():
    # 实际应校验当前用户权限，这里演示挂载位置
    return {"admin": True}

users = APIRouter(prefix="/users", tags=["用户"])

@users.get("/{user_id}")
def detail(user_id: int):
    return {"id": user_id}

@users.delete("/{user_id}", dependencies=[Depends(require_admin)])
def remove(user_id: int):
    return {"deleted": user_id}

只有 DELETE /users/{user_id} 会执行 require_admin。注意：装饰器里的依赖会执行，但结果默认不会成为路由参数；如果路由要使用依赖的返回值，就应在函数参数上写 current_user: Annotated[dict, Depends(require_admin)]。

进阶：用子应用包裹实现真正的模块级中间件

如果确实希望"模块级"也使用真正的中间件（而不是依赖），可以把一个子 FastAPI 应用单独挂中间件，再 mount 到主应用上。

from fastapi import FastAPI, Request

users_app = FastAPI()

@users_app.middleware("http")
async def log_users(request: Request, call_next):
    print("用户模块中间件：进入")
    response = await call_next(request)
    print("用户模块中间件：退出")
    return response

@users_app.get("/users/{user_id}")
def detail(user_id: int):
    return {"id": user_id}

app = FastAPI()
app.mount("/", users_app)

这样 users_app 里的中间件只作用于用户模块。它比依赖实现更"字面"，但会改变路由结构（子应用独立成 ASGI 应用），调试和共享依赖都不如 APIRouter + Depends 直观，因此日常更推荐依赖实现。

三种混合使用与执行顺序

三类可以同时存在。执行顺序是：全局中间件在最外层，模块级依赖先于路由级依赖，路由函数在最里层。

from typing import Annotated
from fastapi import APIRouter, Depends, FastAPI, Header, HTTPException, Request

app = FastAPI()

@app.middleware("http")
async def trace(request: Request, call_next):
    print("全局中间件：进入")
    response = await call_next(request)
    print("全局中间件：退出")
    return response

async def require_tenant(x_tenant: Annotated[str, Header()] = ""):
    print("模块级依赖：require_tenant")
    if not x_tenant:
        raise HTTPException(400, "缺少 X-Tenant 请求头")
    return {"tenant": x_tenant}

async def require_admin():
    print("路由级依赖：require_admin")
    return {"admin": True}

users = APIRouter(prefix="/users", dependencies=[Depends(require_tenant)])

@users.get("/{user_id}", dependencies=[Depends(require_admin)])
def detail(user_id: int):
    print("路由函数：查询用户")
    return {"id": user_id}

app.include_router(users)

请求 GET /users/1（带 X-Tenant 头）时，控制台按顺序打印：

全局中间件：进入
模块级依赖：require_tenant
路由级依赖：require_admin
路由函数：查询用户
全局中间件：退出

相互依赖关系：

• 中间件之间：靠 call_next 串联，后注册的在外层，逐层进出。
• 依赖之间：可以嵌套——一个依赖可以 Depends 另一个依赖，按声明顺序解析（见第 35 篇）。
• 中间件与依赖：中间件总在依赖外层；依赖抛出的 HTTPException 会先经过依赖链，再由中间件在退出阶段拿到最终错误响应。

同一个依赖在同一请求中默认复用计算结果（这是请求内缓存，不是全局缓存）；use_cache=False 可要求重新执行。

实战：全局中间件统一返回格式

很多前端希望后端返回统一结构，例如 {"code": 0, "message": "ok", "data": {...}}。用全局中间件可以在不改动每个接口的情况下统一包装。

import json
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.middleware("http")
async def unified_response(request: Request, call_next):
    response = await call_next(request)

    # 文档相关接口不包装，避免破坏 OpenAPI / Swagger
    path = request.url.path
    if path == "/openapi.json" or path.startswith("/docs") or path.startswith("/redoc"):
        return response

    # 只包装 JSON 响应，文件下载、流式响应等原样返回
    content_type = response.headers.get("content-type", "")
    if not content_type.startswith("application/json"):
        return response

    data = json.loads(response.body)

    # 已经是统一格式就不要再包一层
    if isinstance(data, dict) and "code" in data:
        return response

    ok = response.status_code < 400
    wrapped = {
        "code": 0 if ok else response.status_code,
        "message": "ok" if ok else data.get("detail", "error"),
        "data": data if ok else None,
    }
    return JSONResponse(
        content=wrapped,
        status_code=response.status_code,
        headers=dict(response.headers),
    )

@app.get("/users")
def users():
    return {"id": 1, "name": "示例用户"}

要点：

• 不要一律写 code: 0：错误响应（如 HTTPException）也要根据状态码给出非零 code，让前端能区分成功与失败。
• 避免二次包装：判断响应里是否已有 code 字段，防止重复包裹。
• 放行文档接口：/openapi.json 本身是 JSON，被包装后 Swagger 无法解析；/docs、/redoc 也要排除。
• 只包装 JSON：StreamingResponse、文件下载没有完整 body，不能这样读取，需原样放行。

边界与注意

• 本次请求数据放 request.state，不要放会被并发请求覆盖的全局变量。
• 日志记录方法、路径、状态码和编号；默认不记录密码、令牌及整份上传正文。
• 无条件读取完整正文会增加内存使用，还可能干扰流式请求。
• @app.middleware("http") 针对 HTTP，不会自动覆盖 WebSocket；需要时实现 ASGI 协议层中间件。
• 如果内层抛出未处理异常，await call_next 之后的代码不会运行；生产观测应把结束日志放在 finally，并配合应用异常处理器设计错误响应。

练习：先只注册全局中间件和模块级依赖，观察打印顺序；再加入路由级依赖，验证三种混合后的顺序；给 /users 加一个会抛出 HTTPException 的接口，确认统一返回格式的 code 是否正确反映状态码。