cpu-tts / app /errors.py
grimshaw's picture
Upload current hf-space contents
8986c3d verified
Raw
History Blame Contribute Delete
2.08 kB
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)