在 FastAPI 中全局捕获 `Exception`

Catch `Exception` globally in FastAPI

我正在尝试在全局级别捕获未处理的异常。所以在 main.py 文件的某处我有以下内容:

@app.exception_handler(Exception)
async def exception_callback(request: Request, exc: Exception):
  logger.error(exc.detail)

但是上面的方法一直没有执行。但是,如果我编写一个自定义异常并尝试捕获它(如下所示),它工作正常。

class MyException(Exception):
  #some code

@app.exception_handler(MyException)
async def exception_callback(request: Request, exc: MyException):
  logger.error(exc.detail)

我经历了Catch exception type of Exception and process body request #575。但是这个错误谈论访问请求正文。看到这个bug,感觉应该可以抓到Exception。 我使用的 FastAPI 版本是:fastapi>=0.52.0.

提前致谢:)


更新

答案有多种,在此感谢各位读者和作者。 我在我的应用程序中重新访问了这个解决方案。现在我看到我需要设置 debug=False,默认它是 False,但我在

中将它设置为 True
server = FastAPI(
    title=app_settings.PROJECT_NAME,
    version=app_settings.VERSION,
)

@iedmrc 评论@Kavindu Dodanduwa 给出的答案时,我好像错过了。

首先,我邀请大家熟悉 python 中的异常基础 类。您可以在文档 Built-in Exceptions

中阅读它们

其次,通读 fastApi 默认异常覆盖行为 Override the default exception handlers

您必须了解的是 @app.exception_handler 接受从 Exception 派生的任何异常或子 类。例如 RequestValidationError 是内置于 ValueError 中的 python 的子类,它本身是 Exception.

的子类

所以你必须自己设计异常或者抛出这个背景下可用的异常。我猜你的记录器 logger.error(exc.detail) 出了什么问题,要么没有详细信息字段,要么没有正确的记录器配置。

示例代码:

@app.get("/")
def read_root(response: Response):
    raise ArithmeticError("Divide by zero")


@app.exception_handler(Exception)
async def validation_exception_handler(request, exc):
    print(str(exc))
    return PlainTextResponse("Something went wrong", status_code=400)

输出:

一个标准输出条目和一个带有 Something went wrong

的响应

如果您想捕获所有未处理的异常(内部服务器错误),有一种非常简单的方法。 Documentation

from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import Response

app = FastAPI()

async def catch_exceptions_middleware(request: Request, call_next):
    try:
        return await call_next(request)
    except Exception:
        # you probably want some kind of logging here
        return Response("Internal server error", status_code=500)

app.middleware('http')(catch_exceptions_middleware)

确保将此中间件置于其他一切之前。

你可以这样做。它应该 return 带有自定义错误消息的 json 对象也可以在调试器模式下工作。

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(Exception)
async def validation_exception_handler(request, err):
    base_error_message = f"Failed to execute: {request.method}: {request.url}"
    # Change here to LOGGER
    return JSONResponse(status_code=400, content={"message": f"{base_error_message}. Detail: {err}"})

这是 Fastapi 和 Starlette 上的一个已知问题。

我正在尝试通过以下简单示例全局捕获 StarletteHTTPException。

import uvicorn

from fastapi import FastAPI
from starlette.requests import Request
from starlette.exceptions import HTTPException as StarletteHTTPException
from starlette.responses import JSONResponse

app = FastAPI()


@app.exception_handler(StarletteHTTPException)
async def exception_callback(request: Request, exc: Exception):
    print("test")
    return JSONResponse({"detail": "test_error"}, status_code=500)


if __name__ == "__main__":
    uvicorn.run("test:app", host="0.0.0.0", port=1111, reload=True)


有效。我打开浏览器并调用端点 / 并尝试访问 http://127.0.0.1:1111/ ,它将 return json {"detail":"test_error"} 与 HTTP 代码 "500内部服务器错误”。

但是,当我只在@app.exception_handler,

中将 StarletteHTTPException 更改为 Exception 时
import uvicorn

from fastapi import FastAPI
from starlette.requests import Request
from starlette.exceptions import HTTPException as StarletteHTTPException
from starlette.responses import JSONResponse

app = FastAPI()


@app.exception_handler(Exception)
async def exception_callback(request: Request, exc: Exception):
    print("test")
    return JSONResponse({"detail": "test_error"}, status_code=500)


if __name__ == "__main__":
    uvicorn.run("test:app", host="0.0.0.0", port=1111, reload=True)

当我访问 http://127.0.0.1:1111/ 时,方法 exception_callback 无法捕获 StarletteHTTPException。报404错误。

异常行为应该是:StarletteHTTPException 错误可以被 Exception 修饰的方法 exception_handler 捕获,因为 StarletteHTTPException 是 Exception 的子 class。

但是,这是 Fastapi 和 Starlette 中报告的一个已知问题

所以我们目前无法实现目标。

我找到了一种通过使用中间件来捕获没有“ASGI 应用程序中的异常_”异常的方法。不确定这是否有其他副作用,但对我来说效果很好! @iedmrc

@app.middleware("http")
async def exception_handling(request: Request, call_next):
    try:
        return await call_next(request)
    except Exception as exc:
        log.error("Do some logging here")
        return JSONResponse(status_code=500, content="some content")

添加自定义 APIRoute 也可用于处理全局异常。这种方法的优点是,如果从自定义路由中引发 http 异常,它将由 Starlette 的错误处理程序默认处理:

from typing import Callable

from fastapi import Request, Response, HTTPException, APIRouter, FastAPI
from fastapi.routing import APIRoute
from .logging import logger


class RouteErrorHandler(APIRoute):
    """Custom APIRoute that handles application errors and exceptions"""

    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            try:
                return await original_route_handler(request)
            except Exception as ex:
                if isinstance(ex, HTTPException):
                    raise ex
                logger.exception("uncaught error")
                # wrap error into pretty 500 exception
                raise HTTPException(status_code=500, detail=str(ex))

        return custom_route_handler


router = APIRouter(route_class=RouteErrorHandler)

app = FastAPI()
app.include_router(router)

使用 fastapi==0.68.1 为我工作。

关于自定义路线的更多信息:https://fastapi.tiangolo.com/advanced/custom-request-and-route/