中间件

前言

Web 开发中通常需要一种机制处理请求前和响应后的钩子逻辑，这类函数称为中间件。FastAPI 的中间件在应用处理 HTTP 请求和响应之前或之后执行，可以重写、过滤、修改请求，或对响应进行修改。

HTTP 请求中间件

import time

from fastapi import Request


# 为每个 HTTP 请求计算耗时，并在控制台输出
@app.middleware("http")
async def add_process_time(request: Request, call_next):
    start_time = time.time()
    resp = await call_next(request)
    process_time = time.time() - start_time
    print(f"url: {request.url}, process_time: {process_time:.4f}")
    return resp

CORS 跨域中间件

在前后端分离架构中，前端（如 web.ab.com）访问后端（如 api.ab.com）属于跨域，会触发浏览器同源策略。常见处理方式：

代理：通过同源服务器代理转发请求。
JSONP：只支持 GET，兼容性受限。
CORS：支持的请求方式更多，浏览器兼容性更好，是最常用方案。

FastAPI 提供了 CORSMiddleware：

from fastapi.middleware.cors import CORSMiddleware

origins = [
    "https://api.ab.com",
    "https://api2.ab.com",
    "https://api3.ab.com:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,     # 允许跨域请求的域名列表
    allow_credentials=True,    # 是否允许携带凭证（一般为 cookie）
    allow_methods=["*"],       # 允许的 HTTP 方法
    allow_headers=["*"],       # 允许的请求头
)

注意

生产环境不建议直接 allow_origins=["*"] + allow_credentials=True 组合，应明确列出可信域名。

HTTPSRedirectMiddleware

强制所有请求使用 HTTPS：

from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app.add_middleware(HTTPSRedirectMiddleware)

TrustedHostMiddleware

强制请求头中的 Host 必须来自指定列表：

from fastapi.middleware.trustedhost import TrustedHostMiddleware

allowed_hosts = [
    "example.com",
    "*.example.com",
]
app.add_middleware(TrustedHostMiddleware, allowed_hosts=allowed_hosts)

自定义中间件

基于 BaseHTTPMiddleware

将中间件放到单独模块，例如 middlewares/demo.py：

import time

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware


class TimeCalculate(BaseHTTPMiddleware):
    # dispatch 是必须实现的方法
    async def dispatch(self, request: Request, call_next):
        print(">>> TimeCalculate Middleware <<<")
        start_time = time.time()
        resp = await call_next(request)
        elapsed_time = round(time.time() - start_time, 4)
        print(f"URL: {request.url}, Elapsed time: {elapsed_time}s")
        return resp

引入并注册：

from middlewares import demo

app.add_middleware(demo.TimeCalculate)

带参数的自定义中间件

BaseHTTPMiddleware 子类可以在 __init__ 中接收参数：

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse


class BlackIPMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, denied_ips: tuple[str, ...] = ()):
        super().__init__(app)
        self.denied_ips = denied_ips

    async def dispatch(self, request: Request, call_next):
        if self.denied_ips and request.client:
            if request.client.host in self.denied_ips:
                return JSONResponse({"message": "IP Denied"}, status_code=403)
        return await call_next(request)


app.add_middleware(BlackIPMiddleware, denied_ips=("192.168.1.108",))

备注

更底层的自定义中间件可以直接实现 ASGI 接口（__call__(scope, receive, send)），不继承 BaseHTTPMiddleware，但需要手动处理 scope、receive、send，一般用不到。BaseHTTPMiddleware 的子类必须调用 super().__init__(app)。

基于中间件获取响应内容

日志记录等场景需要在中间件中读取响应体。注意：响应体是流式传输的，需要先读取再重新构造响应：

import json

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response


class ResponseLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        resp: Response = await call_next(request)

        # 读取响应体
        body = b""
        async for chunk in resp.body_iterator:
            body += chunk

        print(f"url={request.url} status={resp.status_code} body={body[:500]}")

        # 必须返回新的 Response，因为原响应的 body_iterator 已被消费
        return Response(
            content=body,
            status_code=resp.status_code,
            headers=dict(resp.headers),
            media_type=resp.media_type,
        )

注意

如果直接返回原来的 resp，客户端会收到空响应体，因为流已被读取。日志里也注意不要记录敏感信息（密码、Token 等）。

中间件内异常处理

在中间件中抛出的异常无法被注册的自定义异常处理器捕获（会统一走顶层的 ServerErrorMiddleware 返回 500），因此中间件内的异常建议直接处理并返回对应响应：

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse


class ExceptionHandleMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        try:
            return await call_next(request)
        except Exception as exc:  # noqa: BLE001
            # 中间件内自行处理异常，返回统一的错误响应
            return JSONResponse(
                status_code=500,
                content={"code": 500, "msg": "Internal Server Error", "detail": str(exc)},
            )


app.add_middleware(ExceptionHandleMiddleware)

提示

业务逻辑（路由函数）里的异常可以正常使用全局异常处理器；只有中间件自身抛出的异常需要这样处理。全局异常处理见异常处理。

前言​

HTTP 请求中间件​

CORS 跨域中间件​

HTTPSRedirectMiddleware​

TrustedHostMiddleware​

自定义中间件​

基于 BaseHTTPMiddleware​

带参数的自定义中间件​

基于中间件获取响应内容​

中间件内异常处理​

前言