FastAPI中如何让中间件在认证依赖后仅运行于已认证端点？

针对你的需求，有两种可行的实现方式，分别适配不同场景：

方案一：将统计逻辑转为依赖（推荐） #

既然你已经通过路由级依赖auth_check保护了端点，直接把统计逻辑做成依赖附加到受保护路由上是最符合FastAPI设计的方式。这种方式天然保证统计逻辑在认证完成后执行，且仅作用于受保护端点，还能直接获取认证后的用户元数据。

步骤1：改写统计逻辑为依赖函数 #

from fastapi import Request, Depends, BackgroundTasks
from sqlalchemy.orm import Session
from your_db_module import get_db  # 替换为你实际的get_db导入路径

async def record_api_usage_dependency(
    request: Request,
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db),
    current_user = Depends(auth_check)  # 直接依赖auth_check，自动获取认证后的用户数据
):
    # 这里可以直接使用current_user的元数据（比如用户ID、角色等）
    background_tasks.add_task(record_api_usage_to_db, db, request, current_user)

步骤2：将统计依赖添加到受保护路由 #

app.include_router(
    admin_router,
    tags=["admin"],
    prefix="/admin",
    dependencies=[Depends(auth_check), Depends(record_api_usage_dependency)],
)

方案二：利用request.state和路由信息改造中间件 #

如果必须使用中间件，可通过以下方式实现：

步骤1：在auth_check中存储认证元数据 #

修改你的认证依赖，将用户信息存入request.state，供后续中间件获取：

async def auth_check(request: Request, db: Session = Depends(get_db)):
    # 你的认证逻辑（比如解析Token、查询用户）
    user = get_current_user_from_token(db, request.headers.get("Authorization"))
    # 将用户信息存入request.state
    request.state.current_user = user
    return user

步骤2：改造中间件逻辑 #

去掉中间件中的Depends（BaseHTTPMiddleware不支持依赖注入），手动处理数据库会话和后台任务，并通过路由信息过滤受保护端点：

from sqlalchemy.orm import Session
from your_db_module import get_db
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.background import BackgroundTasks

async def record_api_usage_to_db_middleware(request: Request, call_next):
    # 先执行路由处理，确保auth_check已完成
    response = await call_next(request)
    
    # 1. 判断当前路由是否为受保护端点（两种方式选其一）
    # 方式A：通过URL前缀判断（简单直接）
    if request.url.path.startswith("/admin"):
        # 2. 检查是否已完成认证
        if hasattr(request.state, "current_user"):
            # 手动获取数据库会话
            db = next(get_db())
            # 创建后台任务并添加到响应中
            background_tasks = BackgroundTasks()
            background_tasks.add_task(
                record_api_usage_to_db, 
                db, 
                request, 
                request.state.current_user  # 传入认证后的用户元数据
            )
            response.background = background_tasks
    
    # 方式B：通过路由依赖精准判断（更灵活）
    # route = request.scope.get("route")
    # if route and is_route_protected(route):
    #     # 执行统计逻辑
    
    return response

# 可选：实现精准判断路由是否受保护的辅助函数
def is_route_protected(route):
    # 检查当前路由的直接依赖
    for dep in getattr(route, "dependencies", []):
        if callable(dep.dependency) and dep.dependency.__name__ == auth_check.__name__:
            return True
    # 递归检查子路由（如果是APIRouter）
    if hasattr(route, "routes"):
        for sub_route in route.routes:
            if is_route_protected(sub_route):
                return True
    return False

步骤3：在main.py中添加中间件 #

from middlewares import record_api_usage_to_db_middleware
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()
app.add_middleware(BaseHTTPMiddleware, dispatch=record_api_usage_to_db_middleware)

内容的提问来源于stack exchange，提问作者nb123