FastAPI 中间件完全指南 – wiki基地

作者： /

FastAPI 中间件完全指南：深入理解与实战应用

在构建现代 Web 应用程序时，有许多跨领域的任务需要在请求到达最终处理逻辑之前或响应返回客户端之前执行，例如日志记录、身份验证、性能监控、安全头添加、CORS 处理等。这些任务如果分散在每个路由处理函数中，会导致大量重复代码，难以维护。这就是“中间件”（Middleware）发挥作用的地方。

FastAPI 作为一款高性能的 Python Web 框架，充分利用了 Starlette 的强大功能，而 Starlette 的核心设计之一就是其基于 ASGI（Asynchronous Server Gateway Interface）的中间件系统。理解并掌握 FastAPI 中的中间件，是构建健壮、可维护和高效 API 的关键。

本文将带您深入了解 FastAPI 中间件的概念、工作原理、不同类型、常见应用场景以及如何编写自定义中间件，助您成为 FastAPI 中间件的专家。

1. 什么是中间件？（Middleware Concepts）

从概念上讲，中间件是一种软件组件，它位于请求与响应之间，能够拦截并处理进出应用程序核心的请求和响应。您可以将其想象成一个管道或一系列过滤器，每个请求在到达最终路由处理函数（FastAPI 中的路径操作函数）之前，会依次经过这个管道中的所有中间件；而响应在从路由处理函数返回客户端之前，也会反向依次经过这些中间件。

工作流示意图：

客户端请求 (Request)
|
V
--------------------
| Middleware 1 | (拦截、处理请求)
--------------------
|
V
--------------------
| Middleware 2 | (拦截、处理请求)
--------------------
|
V
...
|
V
--------------------
| FastAPI 路由处理函数 | (核心业务逻辑)
--------------------
|
V
...
|
V
--------------------
| Middleware 2 | (拦截、处理响应，执行后处理)
--------------------
|
V
--------------------
| Middleware 1 | (拦截、处理响应，执行后处理)
--------------------
|
V
客户端响应 (Response)

每个中间件都有机会在请求进入下一层之前对其进行预处理，或在下一层（包括最终的路由处理函数）返回响应后对其进行后处理。

2. 为什么在 FastAPI 中使用中间件？

FastAPI 构建在 Starlette 之上，而 Starlette 遵循 ASGI 标准。ASGI 是一种异步 Python Web 接口规范，它定义了应用程序（或框架）与服务器之间的通信方式。中间件在 ASGI 应用中扮演着重要角色，它们本身也是 ASGI 应用，能够包装其他 ASGI 应用。

使用中间件在 FastAPI 中带来诸多好处：

  • 代码解耦与复用： 将跨多个路由的通用逻辑（如认证、日志）从路径操作函数中剥离出来，集中管理。
  • 提高效率： 避免在每个路由函数中重复编写相同的逻辑。
  • 系统化处理： 确保某些操作（如安全头设置、CORS 检查）在所有相关请求上强制执行，减少遗漏的可能性。
  • 灵活性： 可以轻松地添加、删除或调整中间件的顺序，改变应用程序的行为，而无需修改核心业务逻辑。
  • 性能监控： 方便地在请求处理的整个生命周期中测量时间，用于性能分析。

3. FastAPI 中的中间件类型

FastAPI（通过 Starlette）支持添加 ASGI 中间件。这些中间件通常通过 app = FastAPI(middleware=[...]) 的方式在应用实例化时进行配置。

主要分为两大类：

  1. 内置中间件 (Built-in Middleware): Starlette/FastAPI 提供了一些常用的中间件，开箱即用。
  2. 自定义中间件 (Custom Middleware): 您可以根据特定需求编写自己的中间件。

3.1 内置中间件

FastAPI/Starlette 提供了多个实用的内置中间件，可以直接导入并使用：

  • CORSMiddleware: 处理跨域资源共享 (CORS) 请求。
  • GZipMiddleware: 自动压缩响应体，减少传输带宽。
  • HTTPSRedirectMiddleware: 强制将所有 HTTP 请求重定向到 HTTPS。
  • StaticFiles: 用于提供静态文件服务。
  • SessionMiddleware: 基于 Cookie 的会话管理（通常与 SessionBackend 一起使用）。
  • AuthenticationMiddleware: 用于处理认证（通常需要配合自定义认证后端）。
  • RateLimitMiddleware (第三方库 fastapi-limiter 或 starlette-limiter 提供)：用于限制请求频率。

示例：添加 CORS 和 GZip 中间件

“`python
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.gzip import GZipMiddleware

app = FastAPI()

添加 CORS 中间件，允许所有来源、所有方法、所有头部

app.add_middleware(
CORSMiddleware,
allow_origins=[““], # 允许访问的源列表，”” 表示允许所有
allow_credentials=True, # 是否允许携带认证信息（如 Cookie）
allow_methods=[““], # 允许的 HTTP 方法列表，”” 表示允许所有
allow_headers=[““], # 允许的请求头部列表，”” 表示允许所有
)

添加 GZip 中间件，响应体大于指定大小时进行压缩

app.add_middleware(GZipMiddleware, minimum_size=1000) # 响应体大小超过 1000 字节时启用压缩

@app.get(“/”)
async def read_root():
return {“Hello”: “World”}

@app.get(“/large_data”)
async def large_data():
# 一个大于 1000 字节的响应
return {“data”: “a” * 2000}
“`

在上述例子中，我们使用了 app.add_middleware() 方法来添加中间件。您也可以在实例化 FastAPI 时通过 middleware 参数传递一个列表：

“`python
from fastapi import FastAPI
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.gzip import GZipMiddleware

在实例化 FastAPI 时通过 middleware 参数传递

app = FastAPI(
middleware=[
Middleware(
CORSMiddleware,
allow_origins=[““],
allow_credentials=True,
allow_methods=[““],
allow_headers=[“*”],
),
Middleware(GZipMiddleware, minimum_size=1000)
]
)

@app.get(“/”)
async def read_root():
return {“Hello”: “World”}
“`

注意，使用 Middleware 类包装中间件类时，第一个参数是中间件类本身，后续参数是传递给该中间件类的 __init__ 方法的参数。

3.2 自定义中间件

自定义中间件是实现特定于应用程序的跨领域逻辑的关键。自定义 ASGI 中间件通常有两种形式：

  1. 函数式中间件 (Function-based Middleware): 一个 ASGI 应用程序函数。
  2. 类式中间件 (Class-based Middleware): 一个实现了 __init__ 和 __call__ 方法的类。

类式中间件更常见，因为它更容易维护状态（通过 __init__ 传递配置或初始化资源）并且结构更清晰。我们将重点介绍类式中间件。

一个典型的自定义类式 ASGI 中间件需要：

  • 继承或遵循 ASGI 应用程序的规范。
  • 一个 __init__(self, app, **kwargs) 方法，其中 app 是下一个 ASGI 应用程序（可以是另一个中间件或最终的 FastAPI 应用）。
  • 一个 async __call__(self, scope, receive, send) 方法，这是 ASGI 应用程序的入口点。scope 包含请求的连接信息，receive 用于接收请求体数据，send 用于发送响应数据。

然而，在 FastAPI/Starlette 中，我们通常不需要直接操作 scope, receive, send 来处理请求/响应的高级细节（如头部、状态码、请求体、响应体）。Starlette 提供了一个更友好的接口，通过 call_next 函数来简化这个过程。

一个基于 call_next 的自定义中间件类的结构如下：

“`python

customs/my_middleware.py

import time
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
from starlette.types import ASGIApp

class CustomHeaderMiddleware(BaseHTTPMiddleware):
def init(self, app: ASGIApp, custom_header: str):
super().init(app)
self.custom_header = custom_header

async def dispatch(self, request: Request, call_next):
    # 在请求到达路由处理函数之前执行的逻辑
    print(f"Adding custom header: {self.custom_header}")

    # 调用下一个中间件或最终的路由处理函数
    response = await call_next(request)

    # 在响应从路由处理函数返回后执行的逻辑
    response.headers["X-Custom-Header"] = self.custom_header
    print(f"Added header X-Custom-Header with value: {self.custom_header}")

    return response

main.py (使用自定义中间件)

from fastapi import FastAPI
from starlette.middleware import Middleware
from .customs.my_middleware import CustomHeaderMiddleware # 假设文件结构如此

app = FastAPI()

app.add_middleware(
CustomHeaderMiddleware,
custom_header=”MyMiddlewareValue”
)

@app.get(“/”)
async def read_root():
return {“Hello”: “World”}
“`

在这个例子中，我们使用了 starlette.middleware.base.BaseHTTPMiddleware 作为基类。这个基类极大地简化了 HTTP 中间件的编写：

  • 您只需要实现一个 dispatch(self, request: Request, call_next) 异步方法。
  • request 是一个 Starlette Request 对象，提供了方便的方式访问请求信息（头部、方法、URL、请求体等）。
  • call_next 是一个可等待对象，调用它会触发管道中的下一个中间件或最终的路由处理函数，并返回一个 Starlette Response 对象。
  • 您可以修改 request 对象（虽然不常见，且需要小心），或者在调用 call_next 之前执行预处理逻辑。
  • 您可以接收 call_next 返回的 response 对象，并在将其返回之前对其进行后处理（例如修改状态码、头部、甚至响应体）。
  • 最后，您需要返回处理后的 response 对象。

BaseHTTPMiddleware 内部处理了复杂的 ASGI scope, receive, send 交互细节，让您能够以更高级别的 HTTP 请求/响应对象进行操作。这是编写自定义 FastAPI 中间件最推荐的方式。

4. 常见的中间件应用场景与实战

4.1 日志记录中间件

记录每个请求的基本信息（方法、路径、客户端 IP、状态码、响应时间）对于监控和调试非常重要。

“`python

customs/logging_middleware.py

import time
import logging
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
from starlette.types import ASGIApp

配置日志

logging.basicConfig(level=logging.INFO, format=’%(asctime)s – %(levelname)s – %(message)s’)
logger = logging.getLogger(name)

class LoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
start_time = time.time()

    # 在请求到达路由处理函数之前记录请求信息
    logger.info(f"Request: {request.method} {request.url} from {request.client.host}")

    try:
        # 调用下一个中间件或路由处理函数
        response = await call_next(request)
    except Exception as exc:
        # 捕获异常以便记录，然后重新抛出
        logger.error(f"Request failed: {request.method} {request.url} - {exc}", exc_info=True)
        raise exc

    # 在响应从路由处理函数返回后记录响应信息和处理时间
    process_time = time.time() - start_time
    logger.info(f"Response: {response.status_code} for {request.method} {request.url} in {process_time:.4f}s")

    # 您也可以选择在这里添加一个响应头来显示处理时间
    # response.headers["X-Process-Time"] = str(process_time)

    return response

main.py (使用日志中间件)

from fastapi import FastAPI
from .customs.logging_middleware import LoggingMiddleware

app = FastAPI()

app.add_middleware(LoggingMiddleware)

@app.get(“/”)
async def read_root():
return {“Hello”: “World”}

@app.get(“/items/{item_id}”)
async def read_item(item_id: int):
if item_id == 0:
raise ValueError(“Item ID cannot be zero”) # 抛出异常测试日志
return {“item_id”: item_id}
“`

这个日志中间件会在请求进入和响应返回时分别打印日志，记录了请求方法、URL、客户端 IP、响应状态码和处理时间。它还包含了异常处理，确保即使路由函数抛出异常，也能被记录下来。

4.2 认证/授权中间件（基础示例）

虽然更复杂的认证和授权通常使用 FastAPI 的依赖注入系统在路径操作函数级别实现，但对于简单的全局认证检查（例如，检查所有请求是否包含特定的 API Key），中间件可以派上用场。

“`python

customs/auth_middleware.py

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.types import ASGIApp

class SimpleAuthMiddleware(BaseHTTPMiddleware):
def init(self, app: ASGIApp, api_key: str):
super().init(app)
self.api_key = api_key

async def dispatch(self, request: Request, call_next):
    # 排除特定路径（例如 /docs, /openapi.json 等）
    if request.url.path in ["/docs", "/openapi.json", "/redoc"]:
         response = await call_next(request)
         return response

    # 检查头部或查询参数中的 API Key
    provided_key = request.headers.get("x-api-key") or request.query_params.get("api_key")

    if not provided_key or provided_key != self.api_key:
        # 如果认证失败，直接返回 JSON 响应并终止请求链
        return JSONResponse(status_code=401, content={"detail": "Invalid or missing API Key"})

    # 如果认证成功，继续处理请求
    response = await call_next(request)
    return response

main.py (使用认证中间件)

from fastapi import FastAPI
from .customs.auth_middleware import SimpleAuthMiddleware

app = FastAPI()

WARNING: 在生产环境中不要硬编码 API Key，应从安全配置中读取

SECRET_API_KEY = “supersecretapikey123”

app.add_middleware(
SimpleAuthMiddleware,
api_key=SECRET_API_KEY
)

@app.get(“/”)
async def read_root():
return {“message”: “Hello, authenticated World!”}

@app.get(“/public”) # 这个路由也会被中间件拦截，除非在中间件中排除
async def public_route():
return {“message”: “This is a public endpoint (if middleware allows it)”}

测试:

GET / -> 401 Unauthorized

GET /?api_key=supersecretapikey123 -> 200 OK

GET / with Header: X-API-Key: supersecretapikey123 -> 200 OK

“`

这个简单的认证中间件检查请求头或查询参数中的 API Key。如果 API Key 不匹配，它会直接返回一个 401 Unauthorized 的 JSON 响应，请求不会到达 / 路由处理函数。请注意，更复杂的基于 Token、OAuth2 或 OIDC 的认证通常结合依赖注入和数据库查询实现，中间件可能只负责解析 Token 并将其放入请求上下文，而具体的权限检查仍在路径操作函数或其依赖中完成。

4.3 添加安全头部中间件

为了提高应用程序的安全性，通常需要在响应中添加一些标准的安全头部。

“`python

customs/security_headers_middleware.py

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
from starlette.types import ASGIApp

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
response = await call_next(request)

    # 添加或修改安全头部
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains; preload"
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    # Content-Security-Policy (CSP) 通常更复杂，可能需要根据页面动态生成
    # response.headers["Content-Security-Policy"] = "default-src 'self';"

    return response

main.py (使用安全头部中间件)

from fastapi import FastAPI
from .customs.security_headers_middleware import SecurityHeadersMiddleware

app = FastAPI()

app.add_middleware(SecurityHeadersMiddleware)

@app.get(“/”)
async def read_root():
return {“Hello”: “World”}
“`

这个中间件会在所有响应中添加一些常见的安全头部，帮助缓解某些类型的 Web 漏洞。

5. 中间件的顺序至关重要

中间件的执行顺序非常重要。请求流过中间件的顺序与您在 middleware 列表中定义它们的顺序相同，而响应则以相反的顺序流回。

考虑以下场景：

  • 认证中间件 vs. 日志中间件： 如果您想记录所有请求（包括未经认证的尝试），日志中间件应该放在认证中间件之前。这样，日志中间件就能拦截并记录那些被认证中间件拒绝的请求。如果日志中间件在认证中间件之后，那么被认证中间件拒绝的请求将永远不会到达日志中间件，也就不会被记录。
  • CORS 中间件 vs. 认证中间件： 通常 CORS 检查应该发生在认证之前。如果一个跨域请求因为缺少必要的 CORS 头部而被浏览器预检请求（OPTIONS 请求）拒绝，那么它甚至不会携带认证凭据到达后端。让 CORS 中间件先处理，可以确保合法的跨域请求能够继续到认证阶段。
  • GZip 中间件 vs. 其他修改响应体的中间件： 如果您有一个中间件需要读取或修改响应体，它通常应该放在 GZip 中间件之前。GZip 中间件会压缩响应体，这使得其他中间件难以直接操作（除非它们也处理解压缩和重新压缩）。

原则：

  • 处理请求的中间件（如认证、日志记录开始部分）按列表顺序执行。
  • 处理响应的中间件（如添加头部、日志记录结束部分、压缩）按列表逆序执行。
  • 更通用的、应该应用于所有（或几乎所有）请求的中间件（如日志、安全头部）通常放在列表前面。
  • 更具体的、可能终止请求处理的中间件（如认证、限速）通常放在通用中间件之后，但在实际业务逻辑之前。
  • 修改响应体的中间件应仔细考虑与压缩中间件的顺序。

示例：合理的中间件顺序

“`python
from fastapi import FastAPI
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.gzip import GZipMiddleware

假设您有之前定义的 LoggingMiddleware, SimpleAuthMiddleware, SecurityHeadersMiddleware

from .customs.logging_middleware import LoggingMiddleware
from .customs.auth_middleware import SimpleAuthMiddleware
from .customs.security_headers_middleware import SecurityHeadersMiddleware

SECRET_API_KEY = “supersecretapikey123”

app = FastAPI(
middleware=[
# 1. 日志记录: 记录所有请求，无论是否认证
Middleware(LoggingMiddleware),
# 2. CORS: 处理跨域预检请求，允许合法跨域请求继续
Middleware(
CORSMiddleware,
allow_origins=[““], allow_credentials=True,
allow_methods=[““], allow_headers=[“*”]
),
# 3. 认证: 检查 API Key，不通过则提前终止请求
Middleware(SimpleAuthMiddleware, api_key=SECRET_API_KEY),
# 4. 安全头部: 在响应返回前添加安全头部 (这里放在认证后，因为即使认证失败的响应也可能需要这些头部)
Middleware(SecurityHeadersMiddleware),
# 5. GZip: 最后对响应体进行压缩（如果需要修改响应体，这个应该在前面）
Middleware(GZipMiddleware, minimum_size=1000),
# … 其他中间件
]
)

@app.get(“/”)
async def protected_route():
return {“message”: “This is a protected endpoint”}

@app.get(“/public”)
async def public_route():
return {“message”: “This endpoint is public (middleware order allows it)”}
“`

在这个例子中，请求首先经过 LoggingMiddleware 被记录，然后经过 CORSMiddleware 处理 CORS，接着是 SimpleAuthMiddleware 检查 API Key。如果认证通过，请求继续到达 / 路由。路由处理完成后，响应返回，依次经过 GZipMiddleware（压缩），SecurityHeadersMiddleware（添加头部），SimpleAuthMiddleware（如果需要对成功响应做后处理，比如添加认证成功标志，虽然这个例子没有），CORSMiddleware（对响应做后处理，比如添加 Access-Control-Allow-Origin 头），最后回到 LoggingMiddleware（记录响应状态和时间）。

6. Middleware vs. Dependencies vs. Routers

这是 FastAPI 开发中一个常见的困惑点。何时使用中间件？何时使用依赖注入？何时只是用路由器？

  • 中间件 (Middleware):

    • 作用范围广： 作用于 整个应用程序 或通过特定配置作用于 所有路由。它在路由匹配和依赖注入 之前 运行（请求进入阶段），并在响应生成 之后 运行（响应返回阶段）。
    • 用途： 处理横切关注点，如全局认证、全局日志、请求/响应转换（如添加头部、压缩）、CORS、限速等，这些是与具体业务逻辑无关或跨越多个业务逻辑的功能。
    • 限制： 难以访问具体的路径操作参数、请求体（不消耗掉的话）或响应体（需要更复杂的 ASGI 操作或 BaseHTTPMiddleware 的特定技巧）。通常不适合进行精细的角色权限控制。

  • 依赖注入 (Dependencies):

    • 作用范围窄： 作用于 特定的路径操作函数、APIRouter 或 其他依赖。在路由匹配 之后，路径操作函数执行 之前 运行。
    • 用途： 用于获取和验证输入数据、注入共享资源（数据库连接、配置对象）、实现精细的认证和授权逻辑（检查用户角色、资源权限）、分页、缓存等。
    • 优势： 可以轻松访问路径参数、查询参数、请求体，并且返回值可以直接注入到路径操作函数中作为参数使用。与业务逻辑紧密相关的功能通常在这里实现。

  • 路由器 (Routers):

    • 作用范围： 用于组织和分组相关的路径操作函数。它们本身不执行业务逻辑或通用处理，只负责路径匹配和将请求导向正确的处理函数。
    • 用途： 模块化大型 API，将不同领域的端点（如用户、商品、订单）组织到单独的文件和逻辑单元中。

总结：

  • 中间件 关注的是 所有 或 大量 请求/响应的 全局 处理。
  • 依赖注入 关注的是 特定 路由或 一组路由 的 前置条件 和 资源注入。
  • 路由器 关注的是 API 的 结构化组织。

什么时候选择中间件？ 当你需要一个功能应用于你的 API 的 大部分 或 所有 部分，并且它独立于具体的路径参数或请求体内容时（或者可以通过 BaseHTTPMiddleware 以 HTTP 级别处理时），考虑使用中间件。

什么时候选择依赖注入？ 当你需要一个功能作用于 特定的 或 一组 路径操作，需要访问请求的详细信息（如请求体数据），或者需要向路径操作函数注入对象或数据时，使用依赖注入。

7. 编写高质量中间件的最佳实践

  • 单一职责原则： 一个中间件应该只负责一个特定的任务（例如，只负责日志，或者只负责认证）。这使得中间件更容易理解、测试和维护。
  • 保持轻量： 中间件会处理每一个请求，避免在中间件中执行耗时或阻塞的操作。如果需要执行 I/O（如数据库查询、外部 API 调用），请确保它们是异步的（使用 await）。
  • 正确使用 call_next： 始终在 dispatch 方法中调用 await call_next(request)。这是将请求传递给应用程序的其余部分并获取响应的方式。确保无论是否发生异常，都要调用 call_next 或返回一个响应，否则请求链会断裂。通常，使用 try...except...finally 结构或在 await call_next() 之后处理响应是安全的模式。
  • 错误处理： 中间件应该能够优雅地处理错误。如果中间件本身逻辑出错，不应该导致整个应用程序崩溃。如果需要捕获下游错误，确保正确记录并重新抛出，或者返回一个适当的错误响应。
  • 考虑性能： 复杂或低效的中间件会成为性能瓶颈，因为它会影响每个请求。
  • 可配置性： 如果中间件需要外部参数（如上面的 api_key），通过 __init__ 方法使其可配置。
  • 测试： 编写测试用例来验证中间件的行为，特别是边缘情况和错误处理。

8. 总结

FastAPI 的中间件系统（基于 Starlette 和 ASGI）是构建强大、灵活和可维护 API 的重要工具。通过拦截请求和响应，中间件允许您在不污染核心业务逻辑的情况下实现日志、认证、安全、性能监控等跨领域功能。

理解 ASGI 中间件的工作原理（尽管 BaseHTTPMiddleware 简化了它）、掌握内置中间件的使用以及学会编写自定义中间件，是充分发挥 FastAPI 潜力的关键技能。同时，请务必注意中间件的顺序及其与依赖注入、路由器的区别，以便在合适的场景选择合适的工具。

通过本文的详细介绍和示例，希望您对 FastAPI 中间件有了全面的认识，并能在您的项目开发中游刃有余地使用它们。

发表评论

您的邮箱地址不会被公开。 必填项已用 * 标注

滚动至顶部