为了解决该问题,我们需要使用 FastAPI 提供的 request-handling middleware(请求处理中间件),在其上注册一个异常处理程序。可以通过在应用程序中自定义中间件来实现这一点。
以下是一些可能的解决方案:
from fastapi.routing import APIRoute
from fastapi import FastAPI, Request, Response
app = FastAPI()
class CustomRoute(APIRoute):
async def handle(self, request: Request, **kwargs) -> Response:
try:
response = await super().handle(request, **kwargs)
return response
except Exception as exc:
print(f"Unexpected error: {exc}")
error_message = {"error": "An unexpected error occurred."}
return JSONResponse(status_code=500, content=error_message)
@app.get('/')
async def root():
return {"message": "Hello World"}
app.router.route_class = CustomRoute
该代码示例使用自定义路由类 CustomRoute
,继承自 APIRoute
。在这个类中,覆盖了 handle
方法,以捕获任何未处理的异常,然后打印错误消息,并返回一个 500 状态码和包含错误消息的 JSON 对象。
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.routing import APIRoute
app = FastAPI()
class CustomRoute(APIRoute):
pass
async def catch_exceptions_middleware(request: Request, call_next):
try:
response = await call_next(request)
except Exception as exc:
error_message = {"error": "An unexpected error occurred."}
response = JSONResponse(status_code=500, content=error_message)
return response
app.middleware("http")(catch_exceptions_middleware)
@app.get('/')
async def root():
return {"message": "Hello World"}
app.router.route_class = CustomRoute
该示例使用了 catch_exceptions_middleware
中间件,该中间件可以捕获任何未处理的异常,并返回一个 500 状态码和包含错误消息的 JSON 对