当前位置：首页 > news >正文

FastAPI中间件

news 2025/7/27 5:39:33

前言

中间件是一个函数，它的每个请求被特定的路径操作处理之前运行，同时也会在每个响应返回之前运行。

它会接收每一个发送到你的应用程序的请求
然后它可以对这个请求进行操作，或者运行一些必要的代码。
接着它将请求传递给应用程序的其余部分（由某个路径操作处理）。
然后它会接收由应用程序（由某个路径操作生成的）响应。
它可以对这个响应进行操作，或者运行一些必要的代码。
最后返回响应。

总的来看，中间件的执行可以划分为以下几个阶段：

1.请求阶段：中间件按添加顺序依次处理请求。
2.路由处理：请求到达对应路径操作函数。
3.响应阶段：中间件按相反顺序处理响应。

请求 ->[中间件1 -> 中间件2 -> 路由处理 -> 中间件2 -> 中间件1] -> 响应

中间件是包裹在应用程序周围的层，可以拦截所有传入的请求和传出的响应。中间件按添加的顺序执行（请求从外到内，响应从内到外）

其适应场景如下：

全局认证（如 JWT 验证）
请求日志记录
添加自定义响应头
异常处理
限流或缓存

内置中间件

FastAPI 内置了一下常用中间件，例如：CORS 中间件

import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddlewareapp = FastAPI()app.add_middleware(CORSMiddleware,allow_origins=["*"],  # 允许的源allow_methods=["*"],  # 允许的 HTTP 方法allow_headers=["*"]  # 允许的请求头
)@app.get("/")
async def index():return {"Hello World"}if __name__ == '__main__':uvicorn.run(app="main:app", port=8000)

HTTPS 重定向中间件

from fastapi.middleware.httpsredirect import HTTPSRedirectMiddlewareapp.add_middleware(HTTPSRedirectMiddleware)  # 强制所有请求使用 HTTPS

创建中间件

除了内置的，我们还可以自定义中间件。要创建中间件，你需要在函数上方使用装饰器 @app.middleware("http")

中间件函数接收以下参数：

请求（request）。
一个函数 call_next，它会将请求作为参数。
这个函数会将请求传递给对应的路径操作。
然后它返回由对应的路径操作生成的响应。
你可以在返回之前进一步修改响应

import uvicorn
import time
from fastapi import FastAPI, Requestapp = FastAPI()@app.middleware("http")
async def add_process_time_header(request: Request, call_next):start_time = time.perf_counter()response = await call_next(request)process_time = time.perf_counter() - start_timeresponse.headers["X-Process-Time"] = str(process_time)return response@app.get("/")
async def index():return {"message": "Hello World"}if __name__ == '__main__':uvicorn.run(app="main:app", port=8000)

上述代码案例在响应请求头中增加了参数 X-Process-Time，处理请求并生成响应所花费的时间（单位：秒）

在这里插入图片描述

自定一个认证中间件

中间件可以用于通过检查认证头部来保护路由

阻止未授权请求

import uvicorn
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import Responseapp = FastAPI()class AuthMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:api_key = request.headers.get("X-API-KEY")if api_key != "secret-key":return Response("Unauthorized", status_code=401)response = await call_next(request)return responseapp.add_middleware(AuthMiddleware)@app.get("/secret-data")
async def index():return {"message": "You accessed protected data!"}if __name__ == '__main__':uvicorn.run(app="main:app", port=8000)