当请求失败并且在 FastAPI 中引发 HTTPException 时如何添加后台任务？答案

【问题标题】：How to add background tasks when request fails and HTTPException is raised in FastAPI?当请求失败并且在 FastAPI 中引发 HTTPException 时如何添加后台任务？
【发布时间】：2022-11-11 02:09:06
【问题描述】：

当我的 FastAPI 端点使用后台任务发生异常时，我试图生成日志：

from fastapi import BackgroundTasks, FastAPI

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:
        background_tasks.add_task(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"}

但是，不会生成日志，因为根据文档here 和here，后台任务“仅”在执行return 语句后运行。

有什么解决方法吗？谢谢。

【问题讨论】：

您可以尝试使用Starlette background tasks

标签： python logging fastapi starlette

【解决方案1】：

这样做的方法是override the HTTPException error handler，由于exception_handler中没有BackgroundTasks对象，您可以按照Starlette documentation（FastAPI is actually Starlette underneath）中描述的方式将后台任务添加到响应中.下面的例子：

from fastapi import BackgroundTasks, FastAPI, HTTPException, Request
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
from starlette.background import BackgroundTask

app = FastAPI()

def write_notification(message):
    with open('log.txt', 'a') as f:
        f.write(f'{message}'+'
')

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    task = BackgroundTask(write_notification, message=exc.detail)
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code, background=task)
 
@app.get("/{msg}")
def send_notification(msg: str, background_tasks: BackgroundTasks):
    if "hello" in msg:
        raise HTTPException(status_code=500, detail="Something went wrong")

    background_tasks.add_task(write_notification, message="Success")
    return {"message": "Request has been successfully submitted."}

如果您需要add multiple background tasks to a response，则使用：

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    tasks = BackgroundTasks()
    tasks.add_task(write_notification, message=exc.detail)
    tasks.add_task(some_other_function, message="some other message")
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code, background=tasks)

上述方法的一种变体如下（建议here）：

from starlette.background import BackgroundTask

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    response = PlainTextResponse(str(exc.detail), status_code=exc.status_code)
    response.background = BackgroundTask(write_notification, message=exc.detail)
    # or, response.background = tasks (create `tasks` as in the previous code snippet)
    return response

一些可能证明对您的任务有用的参考资料是：this answer，它演示了如何添加 custom exception handlers，以及 this answer，它显示了用于传入请求和传出响应的自定义日志记录系统。

【讨论】：