FastAPI中请求失败抛出HTTPException时如何添加后台任务？

核心原因 #

FastAPI的BackgroundTasks默认仅在端点函数正常返回（执行return语句）后才会调度执行，抛出异常时会中断流程，导致已添加的后台任务无法触发。

可行解决方案 #

方案1：手动执行同步任务（简单直接） #

如果你的后台任务是同步函数（比如示例中的write_notification），可以在抛出异常前直接调用该函数，跳过BackgroundTasks的调度逻辑：

from fastapi import BackgroundTasks, FastAPI, HTTPException

app = FastAPI()

def write_notification(message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"{message}"
        email_file.write(content)

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    if "hello" in email:
        # 直接调用同步任务，而非通过BackgroundTasks添加
        write_notification(message="helloworld")
        raise HTTPException(status_code=500, detail="example error")

    background_tasks.add_task(write_notification, message="hello world.")
    return {"message": "Notification sent in the background"}

方案2：利用全局异常处理器处理日志 #

如果需要统一处理所有异常场景的日志，可以定义全局异常处理器，在处理器中添加后台任务或直接执行日志逻辑：

from fastapi import BackgroundTasks, FastAPI, HTTPException
from fastapi.requests import Request
from fastapi.responses import JSONResponse

app = FastAPI()

def write_notification(message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"{message}"
        email_file.write(content)

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    # 在异常处理器中添加后台任务
    background_tasks = BackgroundTasks()
    background_tasks.add_task(write_notification, message=f"异常日志：{exc.detail}")
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
        background=background_tasks  # 将后台任务绑定到响应对象
    )

@app.post("/send-notification/{email}")
async def send_notification(email: str):
    if "hello" in email:
        raise HTTPException(status_code=500, detail="example error")

    background_tasks = BackgroundTasks()
    background_tasks.add_task(write_notification, message="hello world.")
    return {"message": "Notification sent in the background"}, background_tasks

这里的关键是将BackgroundTasks绑定到异常处理器返回的JSONResponse对象上，FastAPI会在发送响应后执行该任务。

方案3：使用异步任务队列（适合复杂场景） #

如果你的任务是异步或需要更可靠的调度（比如重试、延迟执行），可以使用第三方任务队列（如Celery、RQ），这类工具不受FastAPI端点流程的限制，只要任务被提交就会执行。

比如用Celery的简化示例：

# 先安装依赖：pip install celery redis
from celery import Celery
from fastapi import FastAPI, HTTPException

app = FastAPI()
celery_app = Celery("tasks", broker="redis://localhost:6379/0")

@celery_app.task
def write_notification(message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"{message}"
        email_file.write(content)

@app.post("/send-notification/{email}")
async def send_notification(email: str):
    if "hello" in email:
        # 提交Celery任务，立即进入队列执行
        write_notification.delay(message="helloworld")
        raise HTTPException(status_code=500, detail="example error")

    write_notification.delay(message="hello world.")
    return {"message": "Notification sent in the background"}

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