motia-iii/services/exceptions.py

"""
Custom Exception Classes für BitByLaw Integration

Hierarchie:
- IntegrationError (Base)
  - APIError
    - AdvowareAPIError
      - AdvowareAuthError
      - AdvowareTimeoutError
    - EspoCRMAPIError
      - EspoCRMAuthError
      - EspoCRMTimeoutError
  - SyncError
    - LockAcquisitionError
    - ValidationError
    - ConflictError
  - RetryableError
  - NonRetryableError
"""

from typing import Optional, Dict, Any


class IntegrationError(Exception):
    """Base exception for all integration errors"""
    
    def __init__(self, message: str, details: Optional[Dict[str, Any]] = None):
        super().__init__(message)
        self.message = message
        self.details = details or {}


# ========== API Errors ==========

class APIError(IntegrationError):
    """Base class for all API-related errors"""
    
    def __init__(
        self,
        message: str,
        status_code: Optional[int] = None,
        response_body: Optional[str] = None,
        details: Optional[Dict[str, Any]] = None
    ):
        super().__init__(message, details)
        self.status_code = status_code
        self.response_body = response_body


class AdvowareAPIError(APIError):
    """Advoware API error"""
    pass


class AdvowareAuthError(AdvowareAPIError):
    """Advoware authentication error"""
    pass


class AdvowareTimeoutError(AdvowareAPIError):
    """Advoware API timeout"""
    pass


class EspoCRMAPIError(APIError):
    """EspoCRM API error"""
    pass


class EspoCRMAuthError(EspoCRMAPIError):
    """EspoCRM authentication error"""
    pass


class EspoCRMTimeoutError(EspoCRMAPIError):
    """EspoCRM API timeout"""
    pass


# ========== Sync Errors ==========

class SyncError(IntegrationError):
    """Base class for synchronization errors"""
    pass


class LockAcquisitionError(SyncError):
    """Failed to acquire distributed lock"""
    
    def __init__(self, entity_id: str, lock_key: str, message: Optional[str] = None):
        super().__init__(
            message or f"Could not acquire lock for entity {entity_id}",
            details={"entity_id": entity_id, "lock_key": lock_key}
        )
        self.entity_id = entity_id
        self.lock_key = lock_key


class ValidationError(SyncError):
    """Data validation error"""
    
    def __init__(self, message: str, field: Optional[str] = None, value: Any = None):
        super().__init__(
            message,
            details={"field": field, "value": value}
        )
        self.field = field
        self.value = value


class ConflictError(SyncError):
    """Data conflict during synchronization"""
    
    def __init__(
        self,
        message: str,
        entity_id: str,
        source_system: Optional[str] = None,
        target_system: Optional[str] = None
    ):
        super().__init__(
            message,
            details={
                "entity_id": entity_id,
                "source_system": source_system,
                "target_system": target_system
            }
        )
        self.entity_id = entity_id


# ========== Retry Classification ==========

class RetryableError(IntegrationError):
    """Error that should trigger retry logic"""
    
    def __init__(
        self,
        message: str,
        retry_after_seconds: Optional[int] = None,
        details: Optional[Dict[str, Any]] = None
    ):
        super().__init__(message, details)
        self.retry_after_seconds = retry_after_seconds


class NonRetryableError(IntegrationError):
    """Error that should NOT trigger retry (e.g., validation errors)"""
    pass


# ========== Redis Errors ==========

class RedisError(IntegrationError):
    """Redis connection or operation error"""
    
    def __init__(self, message: str, operation: Optional[str] = None):
        super().__init__(message, details={"operation": operation})
        self.operation = operation


class RedisConnectionError(RedisError):
    """Redis connection failed"""
    pass


# ========== Helper Functions ==========

def is_retryable(error: Exception) -> bool:
    """
    Determine if an error should trigger retry logic.
    
    Args:
        error: Exception to check
        
    Returns:
        True if error is retryable
    """
    if isinstance(error, NonRetryableError):
        return False
    
    if isinstance(error, RetryableError):
        return True
    
    if isinstance(error, (AdvowareTimeoutError, EspoCRMTimeoutError)):
        return True
    
    if isinstance(error, ValidationError):
        return False
    
    # Default: assume retryable for API errors
    if isinstance(error, APIError):
        return True
    
    return False


def get_retry_delay(error: Exception, attempt: int) -> int:
    """
    Calculate retry delay based on error type and attempt number.
    
    Args:
        error: The error that occurred
        attempt: Current retry attempt (0-indexed)
        
    Returns:
        Delay in seconds
    """
    if isinstance(error, RetryableError) and error.retry_after_seconds:
        return error.retry_after_seconds
    
    # Exponential backoff: [1, 5, 15, 60, 240] minutes
    backoff_minutes = [1, 5, 15, 60, 240]
    if attempt < len(backoff_minutes):
        return backoff_minutes[attempt] * 60
    
    return backoff_minutes[-1] * 60