FastAPI 中间件全面解析：构建强大、可扩展的Web服务

在构建现代化的Web服务时，处理核心业务逻辑固然重要，但同样 crucial 的是那些跨越多个请求的通用功能，例如日志记录、身份验证、CORS处理、性能监控等等。这些功能如果散布在各个业务端点中，不仅会造成代码重复，难以维护，还会使核心逻辑变得混乱。

这时，中间件 (Middleware) 应运而生。它提供了一种优雅的方式，将这些横切关注点（cross-cutting concerns）从业务逻辑中剥离出来，集中处理。FastAPI 作为一款高性能的现代Web框架，自然也提供了强大且灵活的中间件支持。

本文将带你全面深入地解析 FastAPI 中间件，从概念到实践，从内置到自定义，帮助你彻底理解并熟练运用中间件来构建更健壮、更易维护的FastAPI应用。

一、什么是中间件？为什么需要它？

1. 中间件的概念

从广义上讲，中间件是一种位于操作系统和应用程序之间的软件，负责管理计算机资源和网络通信。但在Web开发的语境下，中间件通常特指位于 Web服务器 和 应用程序核心处理逻辑（路由/端点） 之间的一层或多层软件组件。

想象一下一个HTTP请求的处理过程：
请求 -> Web服务器 -> 中间件层 -> 应用程序核心逻辑 -> 中间件层 -> Web服务器 -> 响应

中间件就像一个“看门人”或“中转站”，每个请求进来时会先经过它，响应出去时也会经过它。每个中间件组件都可以选择在请求到达核心逻辑之前或之后执行一些操作，甚至可以完全拦截请求或响应。

2. 为什么需要中间件？

如前所述，中间件的主要目的是处理那些与具体业务逻辑无关但对几乎所有请求都重要的通用任务。这带来了几个显而易见的优势：

代码解耦与复用： 将通用逻辑集中实现，避免在每个路由处理函数中重复编写相同的代码。
提高可维护性： 需要修改或更新某个通用功能时，只需修改对应的中间件即可，无需触碰大量业务代码。
清晰的应用结构： 使业务逻辑更加聚焦于核心任务，提升代码可读性。
灵活性与扩展性： 可以根据需求轻松地添加、移除或调整中间件的顺序，以改变请求处理流程。
拦截与修改： 中间件可以在请求/响应过程中检查、修改甚至终止请求或生成响应。

总结来说，中间件是实现“关注点分离”（Separation of Concerns）的重要工具，是构建大型、复杂Web应用不可或缺的一部分。

二、FastAPI 中的中间件

FastAPI 基于 Starlette 构建，而 Starlette 提供了成熟且强大的 ASGI (Asynchronous Server Gateway Interface) 中间件支持。FastAPI 的中间件系统完全继承了 Starlette 的能力。

在 FastAPI 中，中间件通常是一个 ASGI 应用程序，它接受一个 ASGI 应用作为参数，并返回一个新的 ASGI 应用。这个新的应用在处理请求时，会先执行自己的逻辑，然后再调用内部的 ASGI 应用（即下一个中间件或最终的端点处理函数）。这种链式调用形成了所谓的“洋葱模型”或“管道模型”。

1. 洋葱模型（The Onion Model）

理解FastAPI（或Starlette）中间件的关键在于理解洋葱模型。
想象你的应用核心是洋葱的中心，每个中间件是洋葱的一层皮。

请求进入： 请求从最外层中间件开始，一层一层向内传递。
核心处理： 请求到达最内层（通常是你的FastAPI应用，负责路由匹配和调用端点函数）。
响应返回： 核心应用生成响应后，响应从最内层中间件开始，一层一层向外传递，直到最外层，最终返回给客户端。

(图片来源：Starlette官方文档)

这意味着：

位于外部的中间件会先处理请求，后处理响应。
位于内部的中间件会后处理请求，先处理响应。

中间件的注册顺序决定了它在洋葱中的位置。先注册的中间件位于外层，后注册的中间件位于内层。

2. FastAPI 中如何添加中间件

在 FastAPI 应用中添加中间件非常简单，通常通过 app.add_middleware() 方法来实现。

“`python
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import Response
import time

class CustomHeaderMiddleware(BaseHTTPMiddleware):
async def dispatch(
self, request: Request, call_next: RequestResponseEndpoint
) -> Response:
# 在请求到达端点前执行
print(“Request received, adding custom header.”)
# 可以在这里修改请求，但不常见

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

    # 在响应从端点返回后执行
    print("Response generated, adding custom header.")
    response.headers["X-My-Custom-Header"] = "Processed-by-Middleware"

    return response

创建FastAPI应用实例

app = FastAPI()

添加中间件

注意：添加的顺序很重要！先添加的在外层。

app.add_middleware(CustomHeaderMiddleware)

添加一个简单的路由用于测试

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

如何运行：

1. 安装uvicorn: pip install uvicorn

2. 保存代码为 main.py

3. 运行: uvicorn main:app –reload

4. 访问 http://127.0.0.1:8000/ 查看响应头

“`

在上面的例子中，app.add_middleware(CustomHeaderMiddleware) 将我们自定义的 CustomHeaderMiddleware 添加到了应用中。所有对 / 路由的请求都会先经过这个中间件。

app.add_middleware() 方法接受一个 ASGI 中间件类作为第一个参数，后面可以接受该类的任意关键字参数，用于初始化中间件实例。

三、内置的常用中间件

FastAPI (通过 Starlette) 提供了一些非常有用的内置中间件，可以直接拿来使用：

starlette.middleware.cors.CORSMiddleware: 处理跨域资源共享 (CORS)。这是Web开发中非常常见的需求。
starlette.middleware.gzip.GZipMiddleware: 对响应进行 GZip 压缩，可以减小响应体大小，加快传输速度。
starlette.middleware.httpsredirect.HTTPSRedirectMiddleware: 强制将所有 HTTP 请求重定向到 HTTPS。
starlette.middleware.trustedhost.TrustedHostMiddleware: 阻止 Host 请求头不匹配允许的主机名的请求，有助于防范一些攻击。
starlette.middleware.base.BaseHTTPMiddleware: 这是一个基类，用于方便地编写基于 HTTP 的中间件（就像我们上面的 CustomHeaderMiddleware 例子）。

下面我们来看一个使用 CORSMiddleware 的例子：

“`python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware # 注意导入路径

app = FastAPI()

配置CORS中间件

app.add_middleware(
CORSMiddleware,
allow_origins=[“http://localhost:3000”, “https://my-frontend-domain.com”], # 允许访问的源列表，或使用 “” 允许所有
allow_credentials=True, # 是否允许携带cookies
allow_methods=[““], # 允许的HTTP方法列表，或使用 “” 允许所有
allow_headers=[““], # 允许的请求头列表，或使用 “*” 允许所有
)

@app.get(“/”)
async def read_root():
return {“message”: “This endpoint allows CORS!”}

现在，来自 http://localhost:3000 或 https://my-frontend-domain.com 的前端应用就可以访问这个接口了。

“`

这个例子展示了如何通过 app.add_middleware() 并传递参数来配置内置中间件的行为。

四、编写自定义中间件

虽然内置中间件很方便，但很多时候我们需要实现特定的逻辑。编写自定义中间件是 FastAPI 开发中一项重要的技能。

自定义中间件通常有两种方式：

使用 BaseHTTPMiddleware (推荐用于大多数HTTP相关的中间件)
直接实现 ASGI 接口 (更底层，更灵活，但也更复杂)

1. 使用 `BaseHTTPMiddleware`

这是编写HTTP中间件最简单和推荐的方式。你只需要继承 BaseHTTPMiddleware 并实现 dispatch 异步方法。

“`python
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import Response
import time

class TimingMiddleware(BaseHTTPMiddleware):
async def dispatch(
self, request: Request, call_next: RequestResponseEndpoint
) -> Response:
# 1. 在请求到达端点前执行逻辑
start_time = time.time()
print(f”Request started: {request.method} {request.url}”)

    # 2. 调用链中的下一个中间件或最终的端点处理函数
    # 这一步非常关键，它将请求传递给应用的核心部分，并等待响应返回
    response = await call_next(request)

    # 3. 在响应从端点返回后执行逻辑
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    print(f"Request finished: {request.method} {request.url} - Processed in {process_time:.4f} seconds")

    # 4. 返回响应
    return response

“`

解析 dispatch 方法：

request: Request: 这是当前的请求对象，你可以从中获取请求方法、URL、头信息、请求体等。
call_next: RequestResponseEndpoint: 这是一个异步函数，代表了链中的下一个处理程序（可能是另一个中间件，或者最终的FastAPI路由处理函数）。你必须调用 await call_next(request) 来将请求传递下去，并获取后续处理生成的响应。
-> Response: dispatch 方法必须返回一个 Response 对象。

在 dispatch 方法中你可以：

在 await call_next(request) 之前：
- 检查请求头、URL、方法等。
- 进行身份验证或授权检查。
- 修改请求对象（虽然不常见，但可以通过 request.state 添加信息）。
- 记录请求信息。
- 如果检查不通过，可以直接构建并返回一个 Response 对象，从而阻止请求继续向下传递到端点。
在 await call_next(request) 之后：
- 获取并检查响应对象（状态码、响应头、响应体）。
- 修改响应头或响应体。
- 记录响应信息或处理时间。
- 处理请求过程中可能抛出的异常（通过 try...except 块）。

2. 直接实现 ASGI 接口

这种方式更底层，需要实现一个 ASGI 应用接口。一个 ASGI 应用就是一个接受三个参数的异步可调用对象：scope, receive, send。这种方式通常用于编写与 HTTP 协议本身交互更紧密的中间件，或者需要处理 WebSocket 连接等非 HTTP 请求的情况。

“`python

这个例子更复杂，通常不推荐，除非你真的需要底层的控制

import time
from starlette.types import ASGIApp, Scope, Receive, Send

class SimpleASGIStatsMiddleware:
def init(self, app: ASGIApp):
self.app = app

async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
    if scope['type'] != 'http':
        # 如果不是HTTP请求，直接传递给下一个应用
        await self.app(scope, receive, send)
        return

    # 记录请求开始时间
    start_time = time.time()

    # 定义一个自定义的 send 函数来拦截和修改响应
    async def send_wrapper(message):
        # 在发送响应头时，添加处理时间信息
        if message['type'] == 'http.response.start':
            process_time = time.time() - start_time
            headers = message.get('headers', [])
            headers.append((b'x-process-time-asgi', str(process_time).encode('utf-8')))
            message['headers'] = headers
        await send(message)

    # 将请求传递给下一个应用，并使用我们的 send_wrapper
    await self.app(scope, receive, send_wrapper)

添加到FastAPI应用中

app.add_middleware(SimpleASGIStatsMiddleware)

“`

这种方式更复杂，因为它直接操作 ASGI 原始数据结构 (scope, receive, send)。BaseHTTPMiddleware 内部已经帮你做了很多HTTP相关的封装，所以对于大多数HTTP中间件场景，优先选择 BaseHTTPMiddleware。

五、中间件中的异常处理

在中间件中处理异常是一个常见的需求。例如，如果下游的端点或另一个中间件抛出了异常，你可能希望在某个中间件中捕获它，记录错误，并返回一个统一的错误响应，而不是让服务器直接返回500错误或崩溃。

你可以在 await call_next(request) 调用周围使用标准的 try...except 块来捕获异常。

“`python
from fastapi import FastAPI, HTTPException, Request
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.responses import JSONResponse, Response

class ExceptionHandlingMiddleware(BaseHTTPMiddleware):
async def dispatch(
self, request: Request, call_next: RequestResponseEndpoint
) -> Response:
try:
# 尝试调用链中的下一个处理程序
response = await call_next(request)
return response
except HTTPException as exc:
# 捕获FastAPI或Starlette的HTTPException
print(f”Caught HTTPException: {exc.status_code} – {exc.detail}”)
return JSONResponse(
status_code=exc.status_code,
content={“detail”: exc.detail}
)
except Exception as exc:
# 捕获其他所有未处理的异常
print(f”Caught unexpected exception: {exc}”)
# 可以记录更详细的错误信息
return JSONResponse(
status_code=500,
content={“detail”: “Internal Server Error”}
)

app = FastAPI()
app.add_middleware(ExceptionHandlingMiddleware) # 添加异常处理中间件

@app.get(“/safe”)
async def safe_endpoint():
return {“message”: “This is safe.”}

@app.get(“/error/http”)
async def http_error_endpoint():
raise HTTPException(status_code=400, detail=”Bad Request from endpoint”)

@app.get(“/error/internal”)
async def internal_error_endpoint():
# 模拟一个未处理的内部错误
result = 1 / 0
return {“result”: result}

运行并测试 /safe, /error/http, /error/internal

“`

在这个例子中，ExceptionHandlingMiddleware 捕获了两种类型的异常：HTTPException (FastAPI/Starlette用于返回特定HTTP错误) 和其他所有 Exception。对于前者，它返回带有相应状态码和细节的JSON响应；对于后者，它返回一个通用的500 Internal Server Error。

重要提示： FastAPI 本身也有自己的异常处理系统 (@app.exception_handler)。中间件中的异常处理位于 FastAPi 自身的异常处理之外。如果一个异常被 FastAPI 的异常处理捕获并生成了响应，那么这个响应会正常地回传给中间件。但如果异常没有被 FastAPI 处理（例如在依赖注入之前或之后，或者在特定的情况下），那么中间件的 try...except 块就有机会捕获它。通常，你会在中间件中处理那些你想全局捕获的、FastAPI默认处理机制可能遗漏或你想定制处理方式的异常。对于特定的HTTPException，使用 app.exception_handler 通常是更推荐和惯用的方式。中间件的异常处理更多用于捕获那些可能导致应用崩溃的、更底层的或意外的异常。

六、中间件的顺序

正如洋葱模型所示，中间件的添加顺序至关重要。先添加的中间件会位于外层，后添加的位于内层。

例如：

“`python
app = FastAPI()

中间件 A (最外层)

app.add_middleware(MiddlewareA)

中间件 B (中层)

app.add_middleware(MiddlewareB)

中间件 C (最内层，紧邻路由处理)

app.add_middleware(MiddlewareC)

请求处理流程：

Request -> MiddlewareA -> MiddlewareB -> MiddlewareC -> Route Handler

响应处理流程：

Response <- MiddlewareA <- MiddlewareB <- MiddlewareC <- Route Handler

“`

这意味着：

MiddlewareA 的 dispatch 方法会最先执行（请求阶段）和最后执行（响应阶段）。
MiddlewareC 的 dispatch 方法会最后执行（请求阶段）和最先执行（响应阶段，紧邻路由处理）。

考虑几个常见的顺序场景：

CORS / TrustedHost： 通常放在最外层。这些中间件负责验证请求的来源或Host，如果验证失败，可以直接拒绝请求，而无需将请求传递给内部处理，从而提高效率和安全性。
Authentication / Authorization： 放在 CORS/TrustedHost 之后，但通常在主要的业务逻辑处理之前。它们需要知道请求的来源是合法的（由外层中间件保证），然后检查请求的身份和权限。如果权限不足，同样可以直接返回错误响应。
Logging / Timing： 可以放在外层或内层，取决于你的日志需求。放在外层可以记录所有请求（包括被内层中间件或路由拒绝的）。放在内层可以更准确地记录到达具体路由的请求的处理时间。记录整个请求生命周期（从进入到离开）的计时器通常放在外层。
GZipMiddleware： 通常放在较内层，靠近响应生成的地方，以便压缩最终的响应体。

仔细思考每个中间件的功能及其依赖关系，合理安排它们的顺序是构建健壮应用的关键。

七、中间件的常见用例

中间件的应用场景非常广泛，几乎所有需要全局处理的通用任务都可以考虑使用中间件。以下是一些常见的用例：

日志记录 (Logging)： 记录每个请求的关键信息（方法、路径、用户、处理时间、状态码等）。
身份验证 (Authentication)： 检查请求头（如 Authorization ），验证用户身份，并将认证信息添加到请求对象中，供后续的路由或依赖使用。
授权 (Authorization)： 在验证身份后，检查用户是否有权访问请求的资源。虽然复杂的授权逻辑可能放在路由内部或依赖中更合适，但简单的全局权限检查可以在中间件中完成。
CORS 处理 (Cross-Origin Resource Sharing)： 允许或拒绝来自不同源的请求（如前端应用）。通常使用内置的 CORSMiddleware。
GZip 压缩： 自动压缩响应体，减少带宽使用，提高加载速度。使用内置的 GZipMiddleware。
流量控制/限速 (Rate Limiting)： 限制来自同一IP地址或用户的请求频率，防止滥用或DDoS攻击。这通常需要一个中间件来检查并更新请求计数。
给响应添加通用 Headers： 例如安全相关的 Headers (Content-Security-Policy, Strict-Transport-Security) 或自定义的 Headers (如处理时间)。
Session 管理： 虽然 FastAPI 本身不内置 Session，但可以使用第三方库（如 python-multipart 的 SessionMiddleware）通过中间件实现基于 Cookie 的 Session 管理。
上下文管理/请求追踪： 为每个请求生成一个唯一的 ID，并将其添加到请求对象或日志中，方便跟踪请求在整个系统中的流程。
修改请求/响应体： 虽然不常见，但在特定场景下（如数据转换、加密/解密）中间件可以读取和修改请求体或响应体。但这通常涉及到异步迭代读取 ASGI body，相对复杂。

八、中间件与 FastAPI 依赖注入 (Dependencies) 的区别与协作

理解中间件和依赖注入各自的定位非常重要，它们服务于不同的目的，但可以协作。

中间件： 运行在路由处理函数之前和之后。它处理的是整个请求/响应生命周期中的横切关注点，与具体的路由路径无关。中间件无法直接访问路由函数的参数，也不能直接利用 FastAPI 的依赖注入系统来获取依赖。但它可以修改请求对象（例如通过 request.state 附加信息），这些信息可能被后续的依赖或路由函数访问。
依赖注入 (Dependencies)： 运行在路由处理函数之前（但通常在所有中间件之后）。它专注于为特定的路由处理函数提供所需的输入参数、执行前置校验或获取资源。依赖注入可以非常方便地访问配置、数据库连接、进行认证授权检查，并将结果作为参数传递给路由函数。

何时使用哪个？

全局性、生命周期性、与具体路由无关的逻辑（如日志、计时、CORS、全局异常捕获、通用响应头） → 中间件。
特定路由需要的数据、资源或前置条件检查（如获取当前用户、验证请求体、分页参数、数据库会话、特定角色的权限检查） → 依赖注入。

它们可以协作：一个中间件可以负责通用的认证（例如，从请求头获取 token 并验证其格式），然后将解析出的用户信息（如用户 ID）存储在 request.state 中。后续的依赖（用于特定路由）可以从 request.state 中取出用户 ID，再进行更细致的授权检查或加载完整的用户对象。

“`python

示例：中间件处理通用认证，依赖注入获取用户

from fastapi import FastAPI, Request, Depends, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.responses import JSONResponse
import base64

简单的模拟用户数据库

FAKE_USERS_DB = {
“user1”: {“username”: “user1”, “role”: “user”},
“admin1”: {“username”: “admin1”, “role”: “admin”},
}

class AuthenticationMiddleware(BaseHTTPMiddleware):
async def dispatch(
self, request: Request, call_next: RequestResponseEndpoint
) -> Response:
# 假设从 Authorization: Basic username:password 获取用户
auth_header = request.headers.get(“Authorization”)
if auth_header and auth_header.startswith(“Basic “):
encoded_credentials = auth_header.split(” “)[1]
try:
decoded_credentials = base64.b64decode(encoded_credentials).decode(“utf-8”)
username, password = decoded_credentials.split(“:”, 1)
# 简单模拟验证（实际应用中会有更安全的密码验证）
if username in FAKE_USERS_DB:
# 将用户信息添加到请求的状态中，供后续依赖使用
request.state.user = FAKE_USERS_DB[username]
print(f”Middleware: Authenticated user: {username}”)
else:
print(f”Middleware: Authentication failed for user: {username}”)
request.state.user = None # 认证失败
except Exception as e:
print(f”Middleware: Authentication header parsing failed: {e}”)
request.state.user = None # 解析失败

    response = await call_next(request)
    return response

依赖函数：从请求状态中获取用户

def get_current_user(request: Request):
user = request.state.user
if user is None:
# 如果中间件没有成功认证用户，依赖抛出异常
raise HTTPException(status_code=401, detail=”Not authenticated”)
return user

依赖函数：检查用户是否是管理员

def get_admin_user(current_user: dict = Depends(get_current_user)):
if current_user.get(“role”) != “admin”:
raise HTTPException(status_code=403, detail=”Operation not permitted”)
return current_user

app = FastAPI()
app.add_middleware(AuthenticationMiddleware) # 添加认证中间件

@app.get(“/users/me”)
async def read_users_me(current_user: dict = Depends(get_current_user)):
# 依赖 get_current_user 会先运行，确保用户已认证
return {“username”: current_user[“username”], “role”: current_user[“role”]}

@app.get(“/admin”)
async def read_admin_info(admin_user: dict = Depends(get_admin_user)):
# 依赖 get_admin_user 会先运行，确保用户是管理员
return {“message”: f”Welcome admin: {admin_user[‘username’]}”}

运行并测试：

GET /users/me with Basic YWRtaW4xOnBhc3N3b3Jk -> Success (user/admin)

GET /users/me without auth header -> 401

GET /admin with Basic YWRtaW4xOnBhc3N3b3Jk -> Success (admin)

GET /admin with Basic dXNlcjE6cGFzc3dvcmQ= -> 403 (user)

“`

这个例子清晰地展示了中间件（处理全局认证逻辑）和依赖注入（获取认证结果并用于特定路由的授权检查）如何协同工作。

九、总结

FastAPI 中间件是一个强大且灵活的工具，用于处理Web应用中的横切关注点。通过理解其洋葱模型和 dispatch 方法的工作原理，我们可以轻松地实现各种功能，如日志、认证、CORS、性能监控等。

BaseHTTPMiddleware 提供了一种简洁的方式来编写基于 HTTP 的中间件，满足了大多数需求。对于更底层或非 HTTP 的场景，可以直接实现 ASGI 接口。合理地组织和排序中间件，并结合 FastAPI 强大的依赖注入系统，可以帮助我们构建出结构清晰、易于维护、高性能的现代化Web服务。

熟练掌握 FastAPI 中间件的使用，将极大地提升你的开发效率和应用质量。现在，是时候将这些知识应用到你的 FastAPI 项目中，让你的应用更上一层楼了！

FastAPI 中间件 全面解析：构建强大、可扩展的Web服务