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

Question

当 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"}

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

有什么解决方法吗？

Answer 1

实现这一点的方法是重写HTTPException错误处理程序BackgroundTasks，并且由于中没有对象，因此您可以按照Starlette 文档exception_handler中描述的方式向响应添加后台任务（FastAPI 实际上是下面的 Starlette）。

顺便说一句，FastAPI 将运行async def直接在事件循环中定义的后台任务，而使用正常定义的后台任务def将在与外部线程池分开的线程中运行await，然后进行编辑，否则会阻塞事件循环。它与 API 端点的概念相同。请查看此答案和此答案以获取更多详细信息，这将帮助您决定如何定义后台任务函数。

例子

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}'+'\n')

@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."}

如果您需要向响应添加多个后台任务，请使用：

from fastapi import BackgroundTasks

@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)

上述方法的变体如下（此处建议）：

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)
    # to add multiple background tasks use:
    # response.background = tasks  # create `tasks` as shown in the code above
    return response  

以下是一些可能对您有所帮助的参考资料：（1）此答案演示了如何添加自定义异常处理程序，（2）此答案显示了传入请求和传出响应的自定义日志记录系统。