FastAPI中Starlette中间件异步警告捕获的请求混淆问题咨询

问题根源：中间件方案的并发缺陷 #

warnings模块的状态是全局共享的，而FastAPI的异步中间件在处理并发请求时，多个请求会共享同一个警告捕获上下文——当call_next异步执行时，不同请求的警告流会交叉，导致警告串到其他请求里。用asyncio.Lock虽然能解决，但会把异步服务变成串行，完全浪费了FastAPI的异步性能。

装饰器方案的合理性与隐患规避 #

你用装饰器包装端点的思路是对的，因为每个请求进入端点时都会创建独立的warnings.catch_warnings上下文，天然隔离了不同请求的警告，不会出现并发混淆。只要注意以下几点，就能避免隐患：

1. 同时兼容同步和异步端点（FastAPI支持两种类型的路由函数）
2. 不破坏原响应的结构和状态
3. 避免内存泄漏（每次请求都用新的存储容器保存警告）

装饰器实现代码 #

import warnings
import inspect
from typing import Callable
from fastapi import Request, Response
from fastapi.responses import JSONResponse

def capture_warnings(func: Callable) -> Callable:
    # 异步端点处理逻辑
    async def async_handler(request: Request, *args, **kwargs):
        with warnings.catch_warnings(record=True) as captured:
            warnings.simplefilter("always")
            response = await func(request, *args, **kwargs)
        
        return _attach_warnings_to_response(response, captured)
    
    # 同步端点处理逻辑
    def sync_handler(request: Request, *args, **kwargs):
        with warnings.catch_warnings(record=True) as captured:
            warnings.simplefilter("always")
            response = func(request, *args, **kwargs)
        
        return _attach_warnings_to_response(response, captured)
    
    return async_handler if inspect.iscoroutinefunction(func) else sync_handler

def _attach_warnings_to_response(response: Response, captured_warnings):
    # 处理JSON响应：添加warnings字段
    if isinstance(response, JSONResponse):
        data = response.dict()["content"]
        data["warnings"] = [
            {"message": str(warn.message), "category": warn.category.__name__}
            for warn in captured_warnings
        ]
        return JSONResponse(
            content=data,
            status_code=response.status_code,
            headers=response.headers
        )
    # 处理非JSON响应：添加X-API-Warnings响应头
    else:
        if captured_warnings:
            warning_str = "; ".join(
                f"{warn.category.__name__}: {warn.message}"
                for warn in captured_warnings
            )
            headers = response.headers.copy()
            headers["X-API-Warnings"] = warning_str
            return Response(
                content=response.body,
                status_code=response.status_code,
                headers=headers,
                media_type=response.media_type
            )
    return response

装饰器使用示例 #

from fastapi import FastAPI

app = FastAPI()

@app.get("/warnings")
@capture_warnings
async def return_with_warnings(request: Request):
    warnings.warn("数据格式即将废弃，请使用v2接口", DeprecationWarning)
    warnings.warn("当前请求的超时时间设置过短", UserWarning)
    return {"data": "处理完成"}

@app.get("/no-warnings")
@capture_warnings
def return_without_warnings(request: Request):
    return {"data": "无警告处理完成"}

更适合大型API的全局方案：自定义APIRoute #

如果你的API有大量路由，不想逐个加装饰器，可以自定义APIRoute类，自动为所有路由添加警告捕获逻辑，实现全局统一处理：

自定义APIRoute实现 #

import warnings
from fastapi import APIRoute, Request, Response
from fastapi.responses import JSONResponse

class WarningCaptureRoute(APIRoute):
    def get_route_handler(self):
        original_handler = super().get_route_handler()
        
        async def route_handler(request: Request) -> Response:
            with warnings.catch_warnings(record=True) as captured:
                warnings.simplefilter("always")
                response = await original_handler(request)
            
            return _attach_warnings_to_response(response, captured)
        
        return route_handler

# 复用之前的响应处理函数
def _attach_warnings_to_response(response: Response, captured_warnings):
    if isinstance(response, JSONResponse):
        data = response.dict()["content"]
        data["warnings"] = [
            {"message": str(warn.message), "category": warn.category.__name__}
            for warn in captured_warnings
        ]
        return JSONResponse(
            content=data,
            status_code=response.status_code,
            headers=response.headers
        )
    else:
        if captured_warnings:
            warning_str = "; ".join(
                f"{warn.category.__name__}: {warn.message}"
                for warn in captured_warnings
            )
            headers = response.headers.copy()
            headers["X-API-Warnings"] = warning_str
            return Response(
                content=response.body,
                status_code=response.status_code,
                headers=headers,
                media_type=response.media_type
            )
    return response

全局应用自定义路由 #

app = FastAPI()
# 替换默认路由类为自定义类
app.router.route_class = WarningCaptureRoute

@app.get("/warnings")
async def global_warning_example():
    warnings.warn("全局路由捕获的警告", UserWarning)
    return {"data": "全局处理结果"}

@app.get("/clean")
def clean_example():
    return {"data": "无警告结果"}

核心优势与注意事项 #

• 并发安全：每个请求的警告捕获上下文完全独立，无需锁，不影响FastAPI的异步性能
• 解耦业务与API逻辑：业务代码只需要正常调用warnings.warn()，无需关心警告如何传递给客户端
• 响应兼容性：JSON响应添加字段，非JSON响应添加响应头，不破坏原有响应结构
• 灵活过滤：可以修改warnings.simplefilter规则，只捕获特定类别的警告（比如只处理DeprecationWarning）

内容的提问来源于stack exchange，提问作者Richard Nicolaas Mejerink