| """Platform-level exception hierarchy.""" |
|
|
| from enum import StrEnum |
|
|
|
|
| class ErrorKind(StrEnum): |
| VALIDATION = "invalid_request_error" |
| AUTHENTICATION = "authentication_error" |
| RATE_LIMIT = "rate_limit_exceeded" |
| UPSTREAM = "upstream_error" |
| SERVER = "server_error" |
|
|
|
|
| class AppError(Exception): |
| """Base exception for all application errors.""" |
|
|
| def __init__( |
| self, |
| message: str, |
| *, |
| kind: ErrorKind = ErrorKind.SERVER, |
| code: str = "internal_error", |
| status: int = 500, |
| details: dict | None = None, |
| ) -> None: |
| super().__init__(message) |
| self.message = message |
| self.kind = kind |
| self.code = code |
| self.status = status |
| self.details = details or {} |
|
|
| def to_dict(self) -> dict: |
| err = { |
| "message": self.message, |
| "type": self.kind, |
| "code": self.code, |
| } |
| if "param" in self.details: |
| err["param"] = self.details["param"] |
| return {"error": err} |
|
|
|
|
| class ValidationError(AppError): |
| def __init__(self, message: str, *, param: str = "", code: str = "invalid_value") -> None: |
| super().__init__( |
| message, kind=ErrorKind.VALIDATION, code=code, status=400, |
| details={"param": param}, |
| ) |
| self.param = param |
|
|
|
|
| class AuthError(AppError): |
| def __init__(self, message: str = "Invalid or missing API key") -> None: |
| super().__init__( |
| message, kind=ErrorKind.AUTHENTICATION, code="invalid_api_key", status=401, |
| ) |
|
|
|
|
| class RateLimitError(AppError): |
| def __init__(self, message: str = "No available accounts") -> None: |
| super().__init__( |
| message, kind=ErrorKind.RATE_LIMIT, code="rate_limit_exceeded", status=429, |
| ) |
|
|
|
|
| class UpstreamError(AppError): |
| def __init__( |
| self, |
| message: str, |
| *, |
| status: int = 502, |
| body: str = "", |
| ) -> None: |
| super().__init__( |
| message, kind=ErrorKind.UPSTREAM, code="upstream_error", status=status, |
| details={"body": body}, |
| ) |
|
|
|
|
| class StreamIdleTimeout(AppError): |
| def __init__(self, timeout_s: float) -> None: |
| super().__init__( |
| f"Stream idle timeout after {timeout_s}s", |
| kind=ErrorKind.UPSTREAM, code="stream_idle_timeout", status=504, |
| ) |
|
|
|
|
| __all__ = [ |
| "ErrorKind", "AppError", |
| "ValidationError", "AuthError", "RateLimitError", |
| "UpstreamError", "StreamIdleTimeout", |
| ] |
|
|