🏗️ Exception Hierarchy — Advanced Design Patterns

Master the design of sophisticated exception hierarchies for production systems.

🎨 Designing Exception Hierarchies

Building a well-structured exception hierarchy is crucial for maintainable code:

# Poor: Flat exception structure
class DatabaseError(Exception):
    pass

class NetworkError(Exception):
    pass

class ValidationError(Exception):
    pass

# Better: Hierarchical structure
class AppError(Exception):
    """Base exception for entire application"""
    pass

class PersistenceError(AppError):
    """Data persistence layer errors"""
    pass

class DatabaseError(PersistenceError):
    pass

class ConnectionError(DatabaseError):
    pass

class TransactionError(DatabaseError):
    pass

class DataIntegrityError(DatabaseError):
    pass

class NetworkError(AppError):
    """Network-related errors"""
    pass

class TimeoutError(NetworkError):
    pass

class ConnectionRefusedError(NetworkError):
    pass

class ValidationError(AppError):
    """Data validation errors"""
    pass

class FieldValidationError(ValidationError):
    def __init__(self, field, rule, value):
        self.field = field
        self.rule = rule
        self.value = value
        super().__init__(f"{field}: {rule}")

# Hierarchical catching
def api_call(endpoint):
    try:
        make_request(endpoint)
    except ConnectionRefusedError as e:
        # Handle network connection specifically
        print("Server is not accepting connections")
        raise
    except NetworkError as e:
        # Handle any network error
        print("Network problem occurred")
        raise
    except AppError as e:
        # Handle any application error
        print("Application error occurred")
        raise

🔍 Mixins for Exception Enhancement

# Add behavior to exceptions through mixins
class HTTPStatusMixin:
    """Adds HTTP status code to exceptions"""
    http_status = None

class ErrorCodeMixin:
    """Adds error code for logging"""
    error_code = None

class RetryableMixin:
    """Indicates whether error is retryable"""
    is_retryable = False
    max_retries = 3

# Combine mixins with exception hierarchy
class APIError(HTTPStatusMixin, ErrorCodeMixin, Exception):
    pass

class ClientError(HTTPStatusMixin, APIError):
    http_status = 400

class NotFoundError(ClientError):
    http_status = 404
    error_code = "RESOURCE_NOT_FOUND"

class ValidationError(ClientError):
    http_status = 422
    error_code = "VALIDATION_FAILED"

class ServerError(HTTPStatusMixin, RetryableMixin, APIError):
    http_status = 500
    is_retryable = True

class TemporaryServerError(ServerError):
    http_status = 503
    is_retryable = True
    max_retries = 5

# Usage
def handle_api_error(error):
    if hasattr(error, 'http_status'):
        print(f"HTTP Status: {error.http_status}")
    if hasattr(error, 'error_code'):
        print(f"Error Code: {error.error_code}")
    if hasattr(error, 'is_retryable') and error.is_retryable:
        print(f"Retryable: max {error.max_retries} attempts")

🛡️ Custom Exception Attributes and Methods

# Rich exception with lots of context
class DetailedError(Exception):
    def __init__(self, message, **context):
        self.message = message
        self.context = context
        self.timestamp = time.time()
        self.stacktrace = traceback.format_stack()
        super().__init__(message)

    def __str__(self):
        return f"{self.message} | Context: {self.context}"

    def to_dict(self):
        """Convert exception to dictionary for logging"""
        return {
            "message": self.message,
            "type": self.__class__.__name__,
            "context": self.context,
            "timestamp": self.timestamp
        }

class DataProcessingError(DetailedError):
    def __init__(self, message, input_data=None, processed_so_far=None, **kwargs):
        self.input_data = input_data
        self.processed_so_far = processed_so_far
        super().__init__(message, **kwargs)

    def recovery_suggestion(self):
        """Suggest how to recover"""
        if self.processed_so_far:
            return f"Processed {len(self.processed_so_far)} items before failure"
        return "No items processed before failure"

# Usage
try:
    process_huge_dataset(data)
except DataProcessingError as e:
    print(f"Error: {e}")
    print(f"Recovery: {e.recovery_suggestion()}")
    logger.error(json.dumps(e.to_dict()))

🔄 Exception Translation Layer

# Pattern: Translate framework exceptions to domain exceptions
class UserRepository:
    """Database repository with domain exception translation"""

    def __init__(self, db):
        self.db = db

    def get_by_id(self, user_id):
        """Get user by ID, translating DB exceptions to domain exceptions"""
        try:
            return self.db.query("SELECT * FROM users WHERE id = ?", user_id)
        except db.IntegrityError as e:
            raise DomainError("Data integrity violation") from e
        except db.ConnectionError as e:
            raise RepositoryUnavailable("Cannot connect to database") from e

class DomainError(Exception):
    """Domain-specific error"""
    pass

class RepositoryUnavailable(DomainError):
    """Data repository is unavailable"""
    pass

# Usage in service layer
class UserService:
    def __init__(self, repository):
        self.repository = repository

    def update_user_email(self, user_id, new_email):
        try:
            user = self.repository.get_by_id(user_id)
            user.email = new_email
            self.repository.save(user)
        except RepositoryUnavailable as e:
            # Service layer knows about domain errors, not DB errors
            logger.error(f"Cannot update user: {e}")
            raise UserServiceUnavailable() from e
        except DomainError as e:
            logger.error(f"Domain error: {e}")
            raise

class UserServiceUnavailable(Exception):
    """User service is unavailable"""
    pass

📊 Exception Filtering and Selective Handling

# Context manager for selective exception handling
from contextlib import contextmanager

@contextmanager
def handle_only(exception_types, handler=None):
    """Handle only specific exceptions"""
    try:
        yield
    except exception_types as e:
        if handler:
            handler(e)
        else:
            raise

# Usage
with handle_only(ValueError):
    x = int("invalid")  # ValueError is silently ignored

# More complex: hierarchical filtering
def operation_with_fallbacks():
    try:
        # Try primary approach
        result = primary_approach()
    except SpecificError as e:
        # Handle and continue
        result = secondary_approach()
    except CriticalError:
        # Re-raise without handling
        raise
    except Exception as e:
        # Catch-all with logging
        logger.error(f"Unexpected error: {e}")
        result = default_result()

    return result

# Class-based context manager
class ExceptionHandler:
    """Handle specific exceptions with custom logic"""
    def __init__(self):
        self.handlers = {}
        self.default_handler = None

    def register(self, exception_type, handler):
        """Register handler for exception type"""
        self.handlers[exception_type] = handler
        return self

    def set_default(self, handler):
        """Set default handler for unregistered exceptions"""
        self.default_handler = handler
        return self

    def handle(self, exception):
        """Handle exception with registered handlers"""
        handler = self.handlers.get(type(exception))
        if handler is None:
            handler = self.default_handler

        if handler:
            return handler(exception)
        raise exception

# Usage
handler = ExceptionHandler()
handler.register(ValueError, lambda e: print(f"Invalid value: {e}"))
handler.register(KeyError, lambda e: print(f"Missing key: {e}"))
handler.set_default(lambda e: print(f"Unknown error: {e}"))

try:
    raise ValueError("test")
except Exception as e:
    handler.handle(e)

🌐 Exception Context Propagation

# Track exception context through call stack
class ContextManager:
    """Manages exception context"""
    _context = {}

    @classmethod
    def set_context(cls, **kwargs):
        cls._context.update(kwargs)

    @classmethod
    def get_context(cls):
        return cls._context.copy()

class ContextualError(Exception):
    """Exception with context"""
    def __init__(self, message):
        self.message = message
        self.context = ContextManager.get_context()
        super().__init__(self.format_message())

    def format_message(self):
        ctx = ', '.join(f"{k}={v}" for k, v in self.context.items())
        return f"{self.message} [{ctx}]"

def process_request(request_id, user_id):
    ContextManager.set_context(request_id=request_id, user_id=user_id)
    try:
        do_something()
    except Exception as e:
        raise ContextualError(f"Failed to process: {e}") from e

try:
    process_request("REQ123", "USER456")
except ContextualError as e:
    print(e)  # Shows: Failed to process: ... [request_id=REQ123, user_id=USER456]

🎯 Exception Categorization for Monitoring

# Categorize exceptions for alerting
class ErrorCategory:
    """Categorizes exceptions for monitoring"""
    CRITICAL = "critical"
    WARNING = "warning"
    INFO = "info"

class CategorizedError(Exception):
    category = ErrorCategory.INFO
    should_alert = False

class CriticalError(CategorizedError):
    category = ErrorCategory.CRITICAL
    should_alert = True

class DatabaseError(CriticalError):
    """Database failures are critical"""
    pass

class RateLimitError(CategorizedError):
    category = ErrorCategory.WARNING
    should_alert = False  # Expected in normal operation

class InvalidInputError(CategorizedError):
    category = ErrorCategory.INFO
    should_alert = False

# Monitoring function
def log_exception_for_monitoring(error):
    error_category = getattr(error, 'category', ErrorCategory.INFO)
    should_alert = getattr(error, 'should_alert', False)

    metrics = {
        "error_type": type(error).__name__,
        "category": error_category,
        "message": str(error),
        "should_alert": should_alert
    }

    if should_alert:
        send_alert(metrics)
    else:
        log_to_metrics(metrics)

# Usage
try:
    db.query("SELECT * FROM users")
except DatabaseError as e:
    log_exception_for_monitoring(e)  # Triggers alert
except RateLimitError as e:
    log_exception_for_monitoring(e)  # No alert

🎯 Key Takeaways

• ✅ Design hierarchies for logical error grouping

• ✅ Use mixins to add behavior to exceptions

• ✅ Translate framework exceptions to domain exceptions

• ✅ Include context and metadata in custom exceptions

• ✅ Categorize exceptions for monitoring and alerting

• ✅ Implement exception filtering for selective handling

• ✅ Propagate context through exception chain