184 lines
6.2 KiB
Python
184 lines
6.2 KiB
Python
"""
|
|
HCFS SDK Exception Classes
|
|
|
|
Comprehensive exception hierarchy for error handling.
|
|
"""
|
|
|
|
from typing import Optional, Dict, Any
|
|
|
|
|
|
class HCFSError(Exception):
|
|
"""Base exception for all HCFS SDK errors."""
|
|
|
|
def __init__(self, message: str, error_code: Optional[str] = None, details: Optional[Dict[str, Any]] = None):
|
|
super().__init__(message)
|
|
self.message = message
|
|
self.error_code = error_code
|
|
self.details = details or {}
|
|
|
|
def __str__(self) -> str:
|
|
if self.error_code:
|
|
return f"[{self.error_code}] {self.message}"
|
|
return self.message
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert exception to dictionary for serialization."""
|
|
return {
|
|
"type": self.__class__.__name__,
|
|
"message": self.message,
|
|
"error_code": self.error_code,
|
|
"details": self.details
|
|
}
|
|
|
|
|
|
class HCFSConnectionError(HCFSError):
|
|
"""Raised when connection to HCFS API fails."""
|
|
|
|
def __init__(self, message: str = "Failed to connect to HCFS API", **kwargs):
|
|
super().__init__(message, error_code="CONNECTION_FAILED", **kwargs)
|
|
|
|
|
|
class HCFSAuthenticationError(HCFSError):
|
|
"""Raised when authentication fails."""
|
|
|
|
def __init__(self, message: str = "Authentication failed", **kwargs):
|
|
super().__init__(message, error_code="AUTH_FAILED", **kwargs)
|
|
|
|
|
|
class HCFSAuthorizationError(HCFSError):
|
|
"""Raised when user lacks permissions for an operation."""
|
|
|
|
def __init__(self, message: str = "Insufficient permissions", **kwargs):
|
|
super().__init__(message, error_code="INSUFFICIENT_PERMISSIONS", **kwargs)
|
|
|
|
|
|
class HCFSNotFoundError(HCFSError):
|
|
"""Raised when a requested resource is not found."""
|
|
|
|
def __init__(self, resource_type: str = "Resource", resource_id: str = "", **kwargs):
|
|
message = f"{resource_type} not found"
|
|
if resource_id:
|
|
message += f": {resource_id}"
|
|
super().__init__(message, error_code="NOT_FOUND", **kwargs)
|
|
|
|
|
|
class HCFSValidationError(HCFSError):
|
|
"""Raised when request validation fails."""
|
|
|
|
def __init__(self, message: str = "Request validation failed", validation_errors: Optional[list] = None, **kwargs):
|
|
super().__init__(message, error_code="VALIDATION_FAILED", **kwargs)
|
|
self.validation_errors = validation_errors or []
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
result = super().to_dict()
|
|
result["validation_errors"] = self.validation_errors
|
|
return result
|
|
|
|
|
|
class HCFSRateLimitError(HCFSError):
|
|
"""Raised when rate limit is exceeded."""
|
|
|
|
def __init__(self, retry_after: Optional[int] = None, **kwargs):
|
|
message = "Rate limit exceeded"
|
|
if retry_after:
|
|
message += f". Retry after {retry_after} seconds"
|
|
super().__init__(message, error_code="RATE_LIMIT_EXCEEDED", **kwargs)
|
|
self.retry_after = retry_after
|
|
|
|
|
|
class HCFSServerError(HCFSError):
|
|
"""Raised for server-side errors (5xx status codes)."""
|
|
|
|
def __init__(self, message: str = "Internal server error", status_code: Optional[int] = None, **kwargs):
|
|
super().__init__(message, error_code="SERVER_ERROR", **kwargs)
|
|
self.status_code = status_code
|
|
|
|
|
|
class HCFSTimeoutError(HCFSError):
|
|
"""Raised when a request times out."""
|
|
|
|
def __init__(self, operation: str = "Request", timeout_seconds: Optional[float] = None, **kwargs):
|
|
message = f"{operation} timed out"
|
|
if timeout_seconds:
|
|
message += f" after {timeout_seconds}s"
|
|
super().__init__(message, error_code="TIMEOUT", **kwargs)
|
|
self.timeout_seconds = timeout_seconds
|
|
|
|
|
|
class HCFSCacheError(HCFSError):
|
|
"""Raised for cache-related errors."""
|
|
|
|
def __init__(self, message: str = "Cache operation failed", **kwargs):
|
|
super().__init__(message, error_code="CACHE_ERROR", **kwargs)
|
|
|
|
|
|
class HCFSBatchError(HCFSError):
|
|
"""Raised for batch operation errors."""
|
|
|
|
def __init__(self, message: str = "Batch operation failed", failed_items: Optional[list] = None, **kwargs):
|
|
super().__init__(message, error_code="BATCH_ERROR", **kwargs)
|
|
self.failed_items = failed_items or []
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
result = super().to_dict()
|
|
result["failed_items"] = self.failed_items
|
|
return result
|
|
|
|
|
|
class HCFSStreamError(HCFSError):
|
|
"""Raised for streaming/WebSocket errors."""
|
|
|
|
def __init__(self, message: str = "Stream operation failed", **kwargs):
|
|
super().__init__(message, error_code="STREAM_ERROR", **kwargs)
|
|
|
|
|
|
class HCFSSearchError(HCFSError):
|
|
"""Raised for search operation errors."""
|
|
|
|
def __init__(self, query: str = "", search_type: str = "", **kwargs):
|
|
message = f"Search failed"
|
|
if search_type:
|
|
message += f" ({search_type})"
|
|
if query:
|
|
message += f": '{query}'"
|
|
super().__init__(message, error_code="SEARCH_ERROR", **kwargs)
|
|
self.query = query
|
|
self.search_type = search_type
|
|
|
|
|
|
def handle_api_error(response) -> None:
|
|
"""
|
|
Convert HTTP response errors to appropriate HCFS exceptions.
|
|
|
|
Args:
|
|
response: HTTP response object
|
|
|
|
Raises:
|
|
Appropriate HCFSError subclass based on status code
|
|
"""
|
|
status_code = response.status_code
|
|
|
|
try:
|
|
error_data = response.json() if response.content else {}
|
|
except Exception:
|
|
error_data = {}
|
|
|
|
error_message = error_data.get("error", "Unknown error")
|
|
error_details = error_data.get("error_details", [])
|
|
|
|
if status_code == 400:
|
|
raise HCFSValidationError(error_message, validation_errors=error_details)
|
|
elif status_code == 401:
|
|
raise HCFSAuthenticationError(error_message)
|
|
elif status_code == 403:
|
|
raise HCFSAuthorizationError(error_message)
|
|
elif status_code == 404:
|
|
raise HCFSNotFoundError("Resource", error_message)
|
|
elif status_code == 429:
|
|
retry_after = response.headers.get("Retry-After")
|
|
retry_after = int(retry_after) if retry_after else None
|
|
raise HCFSRateLimitError(retry_after=retry_after)
|
|
elif 500 <= status_code < 600:
|
|
raise HCFSServerError(error_message, status_code=status_code)
|
|
else:
|
|
raise HCFSError(f"HTTP {status_code}: {error_message}") |