在FastAPI中全局捕获`Exception`

Question

我正在尝试在全局级别捕获未处理的异常。所以在

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)

我已经经历了捕获异常类型并处理主体请求#575。但这个错误涉及访问请求正文。看到这个bug后感觉应该可以catch到

Exception

。我使用的FastAPI版本是：

fastapi>=0.52.0

.

提前致谢:)

更新

答案有很多种，感谢这里的所有读者和作者。我在我的应用程序中重新审视了这个解决方案。现在我发现我需要设置

debug=False

，默认是

False

，但我在

 中将其设置为

True

server = FastAPI(
    title=app_settings.PROJECT_NAME,
    version=app_settings.VERSION,
)

当@iedmrc评论@Kavindu Dodanduwa给出的答案时，我似乎错过了。

Answer 1

如果您想捕获所有未处理的异常（内部服务器错误），有一个非常简单的方法。文档

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

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
        print_exception(e)
        return Response("Internal server error", status_code=500)

app.middleware('http')(catch_exceptions_middleware)

确保将此中间件放在其他所有东西之前。

Answer 2

你可以做这样的事情。它应该返回一个 json 对象，其中包含您的自定义错误消息，该对象也可以在调试器模式下工作。

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(Exception)
async def validation_exception_handler(request: Request, exc: Exception):
    # Change here to Logger
    return JSONResponse(
        status_code=500,
        content={
            "message": (
                f"Failed method {request.method} at URL {request.url}."
                f" Exception message is {exc!r}."
            )
        },
    )

Answer 3

添加自定义

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/

Answer 4

这是 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/ ，它将返回 json {"detail":"test_error"} ，HTTP 代码为 "500 Internal Server Error" 。

但是，当我只在@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的子类。

但是，这是 Fastapi 和 Starlette 中报告的已知问题

所以我们目前无法达成目标。

Answer 5

首先我邀请您熟悉Python中的异常基类。您可以在文档中阅读它们内置异常

其次，通读fastApi默认异常覆盖行为覆盖默认异常处理程序

您必须了解的是，

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

Answer 6

4
投票

Something went wrong

Answer 7

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

，然后覆盖

starlette.middleware.exceptions.ExceptionMiddleware

。

这个答案的灵感来自于阅读这个方法：

_lookup_exception_handler()

示例：

starlette.applications.Starlette.build_middleware_stack()

使用示例：

class GenericExceptionMiddleware(ExceptionMiddleware): # Intentional: Defer __init__(...) to super class ExceptionMiddleware # @Override(ExceptionMiddleware) def _lookup_exception_handler( self, exc: Exception ) -> Optional[Callable]: if isinstance(exc, HTTPException): return self.__http_exception_handler else: return self.__exception_handler @classmethod async def __http_exception_handler(cls, request: fastapi.Request, # @Debug ex: HTTPException): log.error("Unexpected error", cause=ex) resp = PlainTextResponse(content=f"Unexpected error: {ex.detail}" f"\n" f"\nException stack trace" f"\n=====================" f"\n{ex}", # Improve to add full stack trace status_code=ex.status_code) return resp @classmethod async def __exception_handler(cls, request: fastapi.Request, # @Debug ex: Exception): log.error("Unexpected error", cause=ex) resp = PlainTextResponse(content=f"Unexpected error: {ex}" f"\n" f"\nException stack trace" f"\n=====================" f"\n{ex}", # Improve to add full stack trace status_code=fastapi.status.HTTP_500_INTERNAL_SERVER_ERROR) return resp

Answer 8

fast_api = FastAPI() fast_api.add_middleware(GenericExceptionMiddleware, debug=fast_api.debug)

在FastAPI中全局捕获`Exception`

问题描述投票：0回答：8

更新

8个回答

最新问题

在FastAPI中全局捕获`Exception`

问题描述 投票：0回答：8

更新

8个回答

最新问题

问题描述投票：0回答：8