不要用这种方法，我之前也是这么写的，但是经常报错，我换了种方式，就是 log 中的 log_msg 作为参数传递过来，不用这个方式

class PermissionException(Exception):
    def __init__(self, body):
        self.body = body

类似于这样，然后在 exc.body 中获取到

终于碰到 fastapi 的人了

fastapi 挺好用的，我们已经有项目应用与生产了。

class PermissionException(Exception):
    def __init__(self, body):
        self.body = body

class DemoException(PermissionException):
    def __init__(self):
        Exception.__init__(self,self.body)
def register_exception(app: FastAPI):
      @user1r(DemoException)
    async def all_exception_handler(request: Request, exc: DemoException):
        # #query参数:
        # query_data = request.query_params
        # print(query_data)
        print(exc.body)
        log_msg = f"捕获到系统错误：请求路径:{request.url.path}\n错误信息：{traceback.format_exc()}"
        mylog.error(log_msg)
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content=jsonable_encoder({
                "responseCode": Status.FAIL.get_code(),
                "responseMsg": Status.FAIL.get_msg()
            },
        ))

用了你的方法还是不行，求解。。。

这时候就无法捕获异常，直接 Internal Server Error，没有进来这个方法。

我这里捕获的是全局报错，不是那种手动抛出异常的。

这段代码是手动捕获异常的处理

# 通用异常处理
class NormalException(Exception):
    def __init__(self, errorMsg: str,body:dict):
        self.errorMsg = str(errorMsg)
        self.body = str(body, encoding = "utf-8")
    # 手动捕获异常处理
    @user1r(NormalException)
    async def unicorn_exception_handler(request: Request, exc: NormalException):
        log_msg = f"手动捕获异常成功====请求路径:{request.url.path}\n请求参数：{exc.body}\n错误信息：{exc.errorMsg}"
        mylog.info(log_msg)
        return JSONResponse(
            status_code=status.HTTP_400_BAD_REQUEST,
            content={
                "responseCode": Status.SYSTEM_EXCEPTION.get_code(),
                "responseMsg":Status.SYSTEM_EXCEPTION.get_msg(),
                "errorMsg" : exc.errorMsg
            },
        )

填坑 (两个解决方案)

博主 - 测试开发坑货（人称无敌哥）刮遍了整个搜索引擎和 fastapi 的 issues，找到了两个解决方案，无敌卷魔牛逼！

自定义中间件-Middleware

在 FastAPI 应用中使用中间件。
中间件实际上是一个函数，在每个 request 处理之前被调用，同时又在每个 response 返回之前被调用。

1. 首先接收访问过来的 request。
2. 然后针对 request 或其他功能执行自定义逻辑。
3. 传递 request 给应用程序继续处理。
4. 接收应用所产生的 response。
5. 然后针对 response 或其他功能执行自定义逻辑。
6. 返回 response。

详细说明可看官方文档：https://fastapi.tiangolo.com/tutorial/middleware/?h=middleware

下面是无敌哥 pity 平台代码，详细可看 git（https://github.com/wuranxu/pity）

async def set_body(request: Request, body: bytes):
    async def receive() -> Message:
        return {"type": "http.request", "body": body}

    request._receive = receive


async def get_body(request: Request) -> bytes:
    body = await request.body()
    await set_body(request, body)
    return body


@pity.middleware("http")
async def errors_handling(request: Request, call_next):
    body = await request.body()
    try:
        await set_body(request, await request.body())
        return await call_next(request)
    except Exception as exc:
        return JSONResponse(
            status_code=status.HTTP_200_OK,
            content=jsonable_encoder({
                "code": 110,
                "msg": str(exc),
                "request_data": body,
            })
        )

参考 issues：https://github.com/tiangolo/fastapi/issues/394
https://stackoverflow.com/questions/61358669/raise-exception-in-python-fastapi-middleware

自定义路由类-APIRoute

在某些情况下，您可能希望覆盖 Request 和 APIRoute 类使用的逻辑。特别是，这可能是中间件中逻辑的一个很好的替代方案。例如，如果您想在应用程序处理请求正文之前读取或操作请求正文。
详细说明可看官方文档：https://fastapi.tiangolo.com/advanced/custom-request-and-route

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

        async def custom_route_handler(request: Request) -> Union[Response]:
            try:
                return await original_route_handler(request)
            except Exception as exc:
                log_msg = f"造数平台捕获到系统错误：请求路径:{request.url.path}\n"
                params = request.query_params
                if params:
                    log_msg += f"路径参数：{params}\n"
                boby = await request.body()
                if boby:
                    body = json.dumps(json.loads(boby),ensure_ascii=False)
                    log_msg +=f"请求参数：{body}\n"
                if isinstance(exc, NormalException):
                    return JSONResponse(
                        status_code=status.HTTP_200_OK,
                        content={
                            "responseCode": Status.SYSTEM_EXCEPTION.get_code(),
                            "responseMsg": exc.errorMsg
                        },
                    )
                elif isinstance(exc, RequestValidationError):
                    message = ""
                    for error in exc.errors():
                        message += str(error.get('loc')[-1]) + ":" + str(error.get("msg")) + ","
                    return JSONResponse(
                        status_code=status.HTTP_200_OK,
                        content=jsonable_encoder({
                            "responseCode": Status.PARAM_ILLEGAL.get_code(),
                            "responseMsg": Status.PARAM_ILLEGAL.get_msg() + message[:-1]
                        })
                    )
                log_msg +=f"错误信息：{str(exc.args[0])}"
                mylog.error(log_msg)
                if PlatConfig.SWITCH == 1:
                    WeGroupChatBot.send_text(log_msg)
                return JSONResponse(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    content=jsonable_encoder({
                        "responseCode": Status.FAIL.get_code(),
                        "responseMsg": Status.FAIL.get_msg(),
                        "errorMsg":str(exc.args[0])
                    },
                    ))
        return custom_route_handler

def APIRouter():
    router = AppRouter()
    router.route_class = ErrorRouter
    return router

统一处理之后，再通过类型判断 Exception，返回不同的 Response~
注意：用了自定义错误路由，就不能再用 @app.exception_handler否则会重复捕获！！！

参考 issues：https://github.com/tiangolo/fastapi/issues/1216
https://github.com/tiangolo/fastapi/issues/2750

无敌哥牛逼

无敌哥牛逼

无敌哥牛逼