| 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) |
|
|