FastAPI中间件屏蔽后台任务异常的原因及解决方法

当添加自定义HTTP中间件后，FastAPI的请求处理流程会经过中间件的上下文。后台任务是在响应返回给客户端之后才执行的，此时中间件的上下文已经结束，Uvicorn顶层的ASGI应用异常捕获器无法再捕获到后台任务抛出的异常，导致异常被静默吞没。

而没有中间件时，后台任务直接在Uvicorn的顶层上下文环境中执行，抛出的异常会被Uvicorn的默认异常处理器捕获并记录到日志中，这是预期的正常行为。

1. 单个任务手动捕获并记录异常 #

在每个后台任务函数内部添加异常捕获，使用Python标准库的logging模块记录完整异常信息：

import logging
from fastapi import BackgroundTasks, FastAPI, Request, Response, status

app = FastAPI(title="Debug demo", description="", version="0.1.0")

@app.middleware("http")
async def noop_middleware(request: Request, call_next):
    return await call_next(request)

@app.post("/test")
async def test(req: Request, background_tasks: BackgroundTasks) -> Response:
    def _fail():
        try:
            raise ValueError("TEST")
        except Exception as e:
            # 记录完整异常栈
            logging.error("后台任务执行失败", exc_info=True)

    background_tasks.add_task(_fail)

    return Response(content="", status_code=status.HTTP_200_OK)

2. 自定义全局后台任务类，统一处理异常 #

通过继承FastAPI的BackgroundTasks类，重写add_task方法，自动为所有后台任务添加异常捕获逻辑，避免重复编写捕获代码：

import logging
from fastapi import BackgroundTasks as FastAPIBackgroundTasks, FastAPI, Request, Response, status

class BackgroundTasks(FastAPIBackgroundTasks):
    def add_task(self, func, *args, **kwargs):
        def wrapped_task(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                logging.error("后台任务执行异常", exc_info=True)
        super().add_task(wrapped_task, *args, **kwargs)

app = FastAPI(title="Debug demo", description="", version="0.1.0")

@app.middleware("http")
async def noop_middleware(request: Request, call_next):
    return await call_next(request)

@app.post("/test")
async def test(req: Request, background_tasks: BackgroundTasks) -> Response:
    def _fail():
        raise ValueError("TEST")

    background_tasks.add_task(_fail)

    return Response(content="", status_code=status.HTTP_200_OK)

3. 全局替换默认BackgroundTasks依赖 #

如果希望所有路由自动使用自定义的后台任务类，可通过FastAPI的依赖注入系统全局替换默认实现：

from fastapi import Depends

# （上述自定义BackgroundTasks类代码）

app.dependency_overrides[FastAPIBackgroundTasks] = lambda: BackgroundTasks()

@app.post("/test")
async def test(req: Request, background_tasks: FastAPIBackgroundTasks) -> Response:
    def _fail():
        raise ValueError("TEST")

    background_tasks.add_task(_fail)

    return Response(content="", status_code=status.HTTP_200_OK)

这样所有路由使用BackgroundTasks时，都会自动应用异常捕获逻辑。

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