DevNull 知识库

Appearance

44｜FastAPI 中间件与请求处理 ​

中间件（Middleware）是处理请求和响应的钩子：在路由函数执行前做认证、在响应返回前加头、在请求结束后记录日志。FastAPI 基于 Starlette 的中间件机制，支持标准的 ASGI 中间件和装饰器式中间件。

一、基础中间件 ​

python
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.time()
        response = await call_next(request)
        duration = time.time() - start
        response.headers["X-Response-Time"] = f"{duration:.3f}s"
        return response

app = FastAPI()
app.add_middleware(TimingMiddleware)

dispatch 接收 Request 和 call_next：

  1. call_next(request) 之前的代码在路由函数执行前运行
  2. call_next(request) 调用下一个中间件或路由函数
  3. call_next(request) 之后的代码在响应生成后运行

二、常用中间件场景 ​

请求 ID 注入 ​

python
import uuid

class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        request.state.request_id = request_id

        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        return response

# 在路由中使用
@app.get("/servers")
async def list_servers(request: Request):
    print(f"处理请求 {request.state.request_id}")
    ...

日志记录 ​

python
import logging

logger = logging.getLogger("access")

class AccessLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        logger.info(
            "%s %s %s",
            request.method,
            request.url.path,
            response.status_code,
        )
        return response

IP 限制 ​

python
class IPWhitelistMiddleware(BaseHTTPMiddleware):
    ALLOWED_IPS = {"127.0.0.1", "10.0.0.0/8"}

    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host
        if client_ip not in self.ALLOWED_IPS:
            from starlette.responses import JSONResponse
            return JSONResponse(
                status_code=403,
                content={"detail": "IP 不在白名单中"},
            )
        return await call_next(request)

三、CORS 中间件 ​

跨域是最常见的中间件需求，FastAPI 内置了 CORS 中间件：

python
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],   # 允许的源
    allow_credentials=True,                      # 允许携带 Cookie
    allow_methods=["*"],                         # 允许所有方法
    allow_headers=["*"],                         # 允许所有头
)

详细内容在下一篇专门讲。

四、Gzip 压缩 ​

python
from fastapi.middleware.gzip import GZipMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)

响应体超过 1000 字节时自动压缩，减少传输体积。

五、中间件执行顺序 ​

中间件按添加顺序执行，像洋葱一样层层包裹：

python
app.add_middleware(MiddlewareA)   # 最外层
app.add_middleware(MiddlewareB)   # 中间层
app.add_middleware(MiddlewareC)   # 最内层

执行顺序：

A 前 → B 前 → C 前 → 路由函数 → C 后 → B 后 → A 后

越先添加的中间件越外层，对请求的处理越早，对响应的处理越晚。

六、常见错误 ​

中间件中不调用 call_next ​

python
class BrokenMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        return JSONResponse({"error": "blocked"})
        # 错误：没有 call_next，路由函数永远不会执行

在中间件中抛异常不处理 ​

python
class RiskyMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        risky_operation()   # 可能抛异常
        return await call_next(request)

中间件抛异常会导致请求直接 500，应该在中间件内部加 try/catch，或确保只执行安全操作。

中间件顺序错误 ​

python
# 错误：Gzip 在 CORS 外层，CORS 头不会被压缩（虽然不影响功能，但逻辑不清晰）
app.add_middleware(GZipMiddleware)
app.add_middleware(CORSMiddleware)

# 推荐：CORS 在最外层（最先处理跨域预检），Gzip 在靠近路由的位置
app.add_middleware(CORSMiddleware)
app.add_middleware(GZipMiddleware)