FastAPI/Starlette：如何处理BackgroundTasks中的异常？

后台任务的异常无法通过Starlette的exception_middleware捕获——因为这类任务是在响应返回给客户端之后才执行的，已经脱离了请求-响应的生命周期。处理后台任务异常的核心思路是在任务内部或统一包装层实现异常捕获与处理，以下是几种实用方案：

1. 直接在任务函数内捕获异常 #

这是最直观的方式，针对单个任务的逻辑做针对性异常处理：

import logging
from fastapi import FastAPI, BackgroundTasks
import uvicorn

# 配置日志
logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)

app = FastAPI()

def test_func(a, b):
    try:
        # 原任务业务逻辑
        raise ValueError("模拟不可预见的异常")
    except Exception as e:
        # 日志记录（exc_info=True会打印完整堆栈信息，方便排查）
        logger.error(f"后台任务test_func执行失败: {str(e)}", exc_info=True)
        # 可扩展逻辑：比如写入错误日志表、发送告警通知等

@app.post("/test", status_code=201)
async def test(background_task: BackgroundTasks):
    background_task.add_task(test_func, 1, 2)
    return {"message": "The test task was successfully sent."}

if __name__ == "__main__":
    uvicorn.run(app=app, host="0.0.0.0", port=8000)

2. 统一异常包装器（避免重复代码） #

如果有多个后台任务，写一个通用包装函数来统一处理异常，减少重复代码：

import logging
from fastapi import FastAPI, BackgroundTasks
import uvicorn

logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)

app = FastAPI()

# 通用后台任务异常包装器
def background_task_wrapper(task_func, *args, **kwargs):
    try:
        return task_func(*args, **kwargs)
    except Exception as e:
        logger.error(f"后台任务{task_func.__name__}执行失败: {str(e)}", exc_info=True)
        # 统一异常处理逻辑：比如发送邮件告警、写入监控系统等

def test_func(a, b):
    raise ValueError("模拟不可预见的异常")

def another_task(c):
    raise IOError("模拟IO异常")

@app.post("/test", status_code=201)
async def test(background_task: BackgroundTasks):
    # 通过包装器添加任务
    background_task.add_task(background_task_wrapper, test_func, 1, 2)
    background_task.add_task(background_task_wrapper, another_task, "test")
    return {"message": "The test task was successfully sent."}

if __name__ == "__main__":
    uvicorn.run(app=app, host="0.0.0.0", port=8000)

3. 自定义BackgroundTasks类（全局控制） #

如果需要全局统一处理所有后台任务的异常，可以继承Starlette的BackgroundTasks类，重写add_task方法：

import logging
from fastapi import FastAPI, BackgroundTasks
from starlette.background import BackgroundTasks as StarletteBackgroundTasks
import uvicorn

logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)

app = FastAPI()

# 自定义BackgroundTasks类，自动添加异常处理
class CustomBackgroundTasks(StarletteBackgroundTasks):
    def add_task(self, func, *args, **kwargs):
        def wrapped_func(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                logger.error(f"后台任务{func.__name__}执行失败: {str(e)}", exc_info=True)
        super().add_task(wrapped_func, *args, **kwargs)

def test_func(a, b):
    raise ValueError("模拟不可预见的异常")

@app.post("/test", status_code=201)
async def test(background_task: CustomBackgroundTasks):
    # 直接使用自定义类添加任务，无需额外包装
    background_task.add_task(test_func, 1, 2)
    return {"message": "The test task was successfully sent."}

if __name__ == "__main__":
    uvicorn.run(app=app, host="0.0.0.0", port=8000)

关键说明 #

• 不需要Starlette的exception_middleware：后台任务执行时请求响应周期已结束，中间件无法捕获其异常。
• 核心原则：确保每个后台任务的异常都被记录或处理，避免静默失败导致问题难以排查。

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