File size: 2,076 Bytes
8986c3d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | from typing import Any
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
class OpenAICompatibleError(Exception):
def __init__(
self,
*,
status_code: int,
message: str,
error_type: str = "invalid_request_error",
param: str | None = None,
code: str | None = None,
) -> None:
self.status_code = status_code
self.message = message
self.error_type = error_type
self.param = param
self.code = code
super().__init__(message)
def error_payload(message: str, error_type: str, param: str | None = None, code: str | None = None) -> dict[str, Any]:
return {
"error": {
"message": message,
"type": error_type,
"param": param,
"code": code,
}
}
async def openai_error_handler(_: Request, exc: OpenAICompatibleError) -> JSONResponse:
return JSONResponse(
status_code=exc.status_code,
content=error_payload(exc.message, exc.error_type, exc.param, exc.code),
)
async def validation_error_handler(_: Request, exc: RequestValidationError) -> JSONResponse:
first = exc.errors()[0] if exc.errors() else {}
loc = first.get("loc", [])
param = loc[-1] if loc else None
msg = first.get("msg", "Invalid request body")
return JSONResponse(
status_code=422,
content=error_payload(msg, "invalid_request_error", str(param) if param else None, None),
)
async def unexpected_error_handler(_: Request, exc: Exception) -> JSONResponse:
return JSONResponse(
status_code=500,
content=error_payload(str(exc) or "Internal server error", "server_error", None, None),
)
def register_exception_handlers(app: FastAPI) -> None:
app.add_exception_handler(OpenAICompatibleError, openai_error_handler)
app.add_exception_handler(RequestValidationError, validation_error_handler)
app.add_exception_handler(Exception, unexpected_error_handler)
|