91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
"""
|
|
Custom exceptions for RunPod workflow operations.
|
|
|
|
All exceptions inherit from RunPodError for easy catching of all module errors.
|
|
"""
|
|
|
|
|
|
class RunPodError(Exception):
|
|
"""Base exception for all RunPod workflow errors."""
|
|
|
|
pass
|
|
|
|
|
|
class ConfigurationError(RunPodError):
|
|
"""Raised when configuration is invalid or missing required values."""
|
|
|
|
def __init__(self, message: str, missing_keys: list[str] | None = None):
|
|
super().__init__(message)
|
|
self.missing_keys = missing_keys or []
|
|
|
|
|
|
class PodDeploymentError(RunPodError):
|
|
"""Raised when pod deployment fails."""
|
|
|
|
def __init__(self, message: str, gpu_type: str | None = None, datacenter: str | None = None):
|
|
super().__init__(message)
|
|
self.gpu_type = gpu_type
|
|
self.datacenter = datacenter
|
|
|
|
|
|
class PodNotFoundError(RunPodError):
|
|
"""Raised when a pod cannot be found."""
|
|
|
|
def __init__(self, pod_id: str):
|
|
super().__init__(f"Pod not found: {pod_id}")
|
|
self.pod_id = pod_id
|
|
|
|
|
|
class PodTerminationError(RunPodError):
|
|
"""Raised when pod termination fails."""
|
|
|
|
def __init__(self, pod_id: str, reason: str):
|
|
super().__init__(f"Failed to terminate pod {pod_id}: {reason}")
|
|
self.pod_id = pod_id
|
|
self.reason = reason
|
|
|
|
|
|
class PodTimeoutError(RunPodError):
|
|
"""Raised when a pod operation times out."""
|
|
|
|
def __init__(self, pod_id: str, operation: str, timeout: int):
|
|
super().__init__(
|
|
f"Pod {pod_id} {operation} operation timed out after {timeout} seconds"
|
|
)
|
|
self.pod_id = pod_id
|
|
self.operation = operation
|
|
self.timeout = timeout
|
|
|
|
|
|
class S3Error(RunPodError):
|
|
"""Raised when S3 operations fail."""
|
|
|
|
def __init__(self, message: str, bucket: str | None = None, key: str | None = None):
|
|
super().__init__(message)
|
|
self.bucket = bucket
|
|
self.key = key
|
|
|
|
|
|
class S3ObjectNotFoundError(S3Error):
|
|
"""Raised when an S3 object doesn't exist."""
|
|
|
|
def __init__(self, bucket: str, key: str):
|
|
super().__init__(f"S3 object not found: s3://{bucket}/{key}", bucket, key)
|
|
|
|
|
|
class APIError(RunPodError):
|
|
"""Raised when RunPod API returns an error."""
|
|
|
|
def __init__(self, message: str, status_code: int | None = None, response: dict | None = None):
|
|
super().__init__(message)
|
|
self.status_code = status_code
|
|
self.response = response
|
|
|
|
|
|
class NetworkError(RunPodError):
|
|
"""Raised when network requests fail."""
|
|
|
|
def __init__(self, message: str, retry_count: int = 0):
|
|
super().__init__(message)
|
|
self.retry_count = retry_count
|