1. 中间件基础结构

from starlette.middleware.base import BaseHTTPMiddlewareclass RequestLoggerMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):# 1. 请求处理前逻辑# 2. 调用后续处理response = await call_next(request)# 3. 响应返回前逻辑return response

2. 简单日志示例

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from logger import loggerclass RequestLoggerMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):if request.method == "GET":logger.info(f"Received request: {request.method} {request.url}")else:body = await request.body()logger.info(f"Received request: {request.method} {request.url}, Body: {body.decode()}")request._body = body  # 关键：恢复bodyresponse = await call_next(request)logger.info(f"Sent response: {response.status_code}")return response

中间件的使用

from fastapi import FastAPI
from middleware.request_logger import RequestLoggerMiddleware
from middleware.token_auth import TokenAuthMiddlewareapp = FastAPI(title="My API")# 添加中间件的顺序很重要！
# 先添加的中间件后执行（对于请求），先执行（对于响应）# 1. 首先添加请求日志中间件（最先记录请求，最后记录响应）
app.add_middleware(RequestLoggerMiddleware)# 2. 然后添加认证中间件
app.add_middleware(TokenAuthMiddleware)# 3. 可以添加其他中间件，如CORS、压缩等
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(CORSMiddleware, allow_origins=["*"])@app.get("/")
async def root():return {"message": "Hello World"}