Spaces:
Sleeping
Sleeping
File size: 2,598 Bytes
bdc2878 | 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 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """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",
]
|