Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| from typing import Any | |
| from fastapi.responses import JSONResponse | |
| def _message_from_value(value: object) -> str: | |
| if isinstance(value, str): | |
| return value | |
| if not isinstance(value, dict): | |
| return "" | |
| message = value.get("message") | |
| if isinstance(message, str) and message: | |
| return message | |
| return _message_from_value(value.get("error")) | |
| def error_message_from_detail(detail: object) -> str: | |
| if isinstance(detail, list): | |
| messages = [] | |
| for item in detail: | |
| if not isinstance(item, dict): | |
| continue | |
| location = ".".join(str(part) for part in item.get("loc", []) if part != "body") | |
| message = str(item.get("msg") or "").strip() | |
| if location and message: | |
| messages.append(f"{location}: {message}") | |
| elif message: | |
| messages.append(message) | |
| return "; ".join(messages) | |
| if isinstance(detail, dict): | |
| message = _message_from_value(detail.get("error")) or _message_from_value(detail) | |
| if message: | |
| return message | |
| return str(detail or "").strip() | |
| def _default_error_type(status_code: int) -> str: | |
| if status_code == 401: | |
| return "authentication_error" | |
| if status_code == 403: | |
| return "permission_error" | |
| if status_code == 429: | |
| return "rate_limit_error" | |
| if 400 <= status_code < 500: | |
| return "invalid_request_error" | |
| return "server_error" | |
| def _default_error_code(status_code: int) -> str: | |
| if status_code == 401: | |
| return "invalid_api_key" | |
| if status_code == 403: | |
| return "permission_denied" | |
| if status_code == 429: | |
| return "rate_limit_exceeded" | |
| if 400 <= status_code < 500: | |
| return "bad_request" | |
| return "upstream_error" | |
| def openai_error_payload( | |
| detail: object, | |
| status_code: int, | |
| *, | |
| error_type: str | None = None, | |
| code: object | None = None, | |
| param: object | None = None, | |
| ) -> dict[str, Any]: | |
| error_detail = detail.get("error") if isinstance(detail, dict) else None | |
| if isinstance(error_detail, dict): | |
| return { | |
| "error": { | |
| "message": error_message_from_detail(error_detail) or "request failed", | |
| "type": str(error_detail.get("type") or error_type or _default_error_type(status_code)), | |
| "param": error_detail.get("param", param), | |
| "code": error_detail.get("code", code if code is not None else _default_error_code(status_code)), | |
| } | |
| } | |
| return { | |
| "error": { | |
| "message": error_message_from_detail(detail) or "request failed", | |
| "type": error_type or _default_error_type(status_code), | |
| "param": param, | |
| "code": code if code is not None else _default_error_code(status_code), | |
| } | |
| } | |
| def openai_error_response( | |
| detail: object, | |
| status_code: int, | |
| *, | |
| headers: dict[str, str] | None = None, | |
| error_type: str | None = None, | |
| code: object | None = None, | |
| param: object | None = None, | |
| ) -> JSONResponse: | |
| return JSONResponse( | |
| status_code=status_code, | |
| content=openai_error_payload(detail, status_code, error_type=error_type, code=code, param=param), | |
| headers=headers, | |
| ) | |
| def anthropic_error_response( | |
| detail: object, | |
| status_code: int, | |
| *, | |
| headers: dict[str, str] | None = None, | |
| ) -> JSONResponse: | |
| error_type = "api_error" if status_code >= 500 else _default_error_type(status_code) | |
| return JSONResponse( | |
| status_code=status_code, | |
| content={ | |
| "type": "error", | |
| "error": { | |
| "type": error_type, | |
| "message": error_message_from_detail(detail) or "request failed", | |
| }, | |
| }, | |
| headers=headers, | |
| ) | |