python-3.x 在FastAPI中全局捕获`Exception`

xtupzzrd  于 2023-03-24  发布在  Python
关注(0)|答案(8)|浏览(588)

我试图在全局级别捕获未处理的异常。所以在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讲的是访问请求体,看到这个bug之后,我觉得应该可以捕获Exception,我现在用的FastAPI版本是:fastapi>=0.52.0 .
提前感谢:)

* 更新 *

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

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

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

w8rqjzmb

w8rqjzmb1#

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

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)

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

up9lanfz

up9lanfz2#

你可以这样做。它应该返回一个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}"})
2sbarzqh

2sbarzqh3#

添加一个自定义的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/

bvuwiixz

bvuwiixz4#

这是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修饰的方法exception_handler捕获StarletteHTTPException错误。
但是,这是Fastapi和Starlette报告的已知问题

所以我们目前还不能达到这个目标。

55ooxyrt

55ooxyrt5#

首先,我邀请大家熟悉一下Python中的异常基类。
其次,通读fastApi默认异常覆盖行为覆盖默认异常处理程序
你必须理解的是@app.exception_handler接受任何Exception或者从Exception派生的子类,例如RequestValidationError是python的一个子类,内置于ValueError中,而ValueError本身是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)

输出:
一个stdout条目和一个带有Something went wrong的响应

3okqufwl

3okqufwl6#

我发现了一种不用“ASGI application_中的异常”就能捕捉异常的方法。不确定这是否有其他副作用,但对我来说效果很好!@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")
cwxwcias

cwxwcias7#

我正在搜索快速API的全局处理程序,用于为我找到并实现的429状态代码提供自定义消息,对于me @app.exception_handler(429)async def ratelimit_handler(请求:请求,不包括:例外):返回JSON响应('消息':“您已超出请求配额。请稍后再试。",'status':'失败'})

2exbekwf

2exbekwf8#

子类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

样品使用:

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

相关问题