FastAPI：如何为特定端点/路径操作添加自定义中间件

问题1：中间件能否仅针对特定请求执行？ #

FastAPI的全局@app.middleware('http')默认会拦截所有HTTP请求，中间件本身一定会被执行，但你可以通过逻辑判断跳过核心业务代码，只对目标请求生效。

你当前代码里获取路径的方式过于繁琐，直接用request.url.path就能拿到请求路径，无需遍历request.items()。优化后的中间件代码如下：

import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware('http')
async def add_process_time_header(request: Request, call_next):
    # 先判断是否为目标请求，非目标请求直接转发，跳过核心逻辑
    if request.url.path != '/users' or request.method != 'GET':
        return await call_next(request)
    
    # 仅对目标请求执行计时逻辑
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers['X-Process-Time'] = str(process_time)
    return response

这种写法下，中间件虽仍会被所有请求触发，但非目标请求会直接跳转，额外性能影响可忽略。

问题2：是否应该用绑定特定请求的依赖项？ #

是的，针对特定端点的逻辑，用依赖项比全局中间件更合适。

依赖项可以精准绑定到单个路径操作、一组路径操作（通过Router）或整个应用，无需在全局做路径过滤。以下是用依赖项实现相同计时逻辑的两种方式：

方式1：单个路径操作绑定依赖 #

import time
from fastapi import FastAPI, Request, Depends
from fastapi.responses import JSONResponse

app = FastAPI()

def record_process_time(request: Request):
    start_time = time.time()
    yield  # yield前为请求处理前逻辑，yield后为请求处理后逻辑
    process_time = time.time() - start_time
    request.state.process_time = process_time

@app.get('/users', dependencies=[Depends(record_process_time)])
async def get_users(request: Request):
    # 业务逻辑示例
    data = {"users": ["Alice", "Bob"]}
    # 将计时结果添加到响应头
    return JSONResponse(content=data, headers={"X-Process-Time": str(request.state.process_time)})

方式2：路由组绑定依赖 #

如果多个端点需要复用同一逻辑，可通过Router统一绑定：

from fastapi import APIRouter, Request, Depends
from fastapi.responses import JSONResponse

router = APIRouter(dependencies=[Depends(record_process_time)])

@router.get('/users')
async def get_users(request: Request):
    data = {"users": ["Alice", "Bob"]}
    return JSONResponse(content=data, headers={"X-Process-Time": str(request.state.process_time)})

@router.get('/users/{user_id}')
async def get_user(request: Request, user_id: int):
    data = {"user_id": user_id, "name": "Alice"}
    return JSONResponse(content=data, headers={"X-Process-Time": str(request.state.process_time)})

app.include_router(router)

生产级设计方案 #

1. 场景化选择方案：
   • 全局通用逻辑（如日志、认证、跨域处理）：使用全局中间件。
   • 特定端点/路由组的逻辑（如计时、权限校验、参数预处理）：优先使用依赖项，代码更清晰，维护成本更低。
2. 路由级中间件：若需在路由级别统一修改请求/响应（如给某个路由组的所有响应加特定头），可给Router单独添加中间件：

   router = APIRouter()
   
   @router.middleware('http')
   async def router_middleware(request: Request, call_next):
       # 仅对该路由组下的请求生效
       start_time = time.time()
       response = await call_next(request)
       process_time = time.time() - start_time
       response.headers['X-Process-Time'] = str(process_time)
       return response
   

3. 避免过度依赖全局中间件：全局中间件的过滤逻辑会增加代码复杂度，排查问题时需覆盖所有请求流转路径；依赖项可精准控制逻辑范围，更易维护。

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