FastAPI/Starlette：如何处理后台任务中的异常？答案

【问题标题】：FastAPI/Starlette: How to handle exceptions inside background tasks?FastAPI/Starlette：如何处理后台任务中的异常？
【发布时间】：2023-01-17 20:40:56
【问题描述】：

我使用 FastAPI 开发了一些 API 端点。这些端点被允许运行BackgroundTasks。不幸的是，我不知道如何处理这些任务中不可预测的问题。

我的 API 示例如下所示：

# main.py

from fastapi import FastAPI
import uvicorn


app = FastAPI()


def test_func(a, b):
    raise ...


@app.post("/test", status_code=201)
async def test(request: Request, background_task: BackgroundTasks):
    background_task.add_task(test_func, a, b)
    return {
        "message": "The test task was successfully sent.",
    }
if __name__ == "__main__":
    uvicorn.run(
        app=app,
        host="0.0.0.0",
        port=8000
    )
# python3 main.py to run
# fastapi == 0.78.0
# uvicorn == 0.16.0

你能帮我处理这种后台任务的任何类型的异常吗？我是否应该添加来自 Starlette 的任何 exception_middleware，以实现此目的？

【问题讨论】：

为什么你不能使用try / except声明？

标签： python exception fastapi background-task starlette

【解决方案1】：

你能帮我处理吗随便哪种后台任务异常？

Background tasks，顾名思义，就是要在后台运行的任务后返回响应。因此，您不能 raise 或 Exception 并期望客户端收到某种响应。如果你只是想赶上任何Exception发生在后台任务中，您可以简单地使用try-except块来捕获异常并根据需要进行处理。例如：
def test_func(a, b):
    try:
        # some background task logic here...
        raise <some_exception>
    except Exception as e:
        print('Something went wrong')
        # use `print(e.detail)` to print out the Exception's details 
如果您想记录任务中出现的任何异常（而不是仅仅将它们打印出来），您可以使用 Python 的 logging 模块——看看 this answer，以及 this answer 和 this answer怎么做。您还可以在 this post 和 this post 以及 here、here 和 here 找到有关 FastAPI/Starlette 的自定义/全局异常处理程序的有用信息。

【讨论】：

【解决方案2】：

我不是 100% 确定你所说的“不可预测的错误”是什么意思，如果发生异常会有什么行为？

try/except 语句可以工作。

# main.py

from fastapi import FastAPI
import uvicorn


app = FastAPI()


def test_func(a, b):
    raise ...


@app.post("/test", status_code=201)
async def test(request: Request, background_task: BackgroundTasks):
    try:
        background_task.add_task(test_func, a, b)
        return {
            "message": "The test task was successfully sent.",
        }
    except Exception as e:
        # exception handling code
if __name__ == "__main__":
    uvicorn.run(
        app=app,
        host="0.0.0.0",
        port=8000
    )
# python3 main.py to run
# fastapi == 0.78.0
# uvicorn == 0.16.0

【讨论】：