🎨 Custom Exceptions — Advanced Enterprise Patterns

Design comprehensive exception systems for production applications.

🏗️ Exception Factories and Builders

# Factory pattern for exception creation
class ExceptionFactory:
    """Creates properly configured exceptions"""

    @staticmethod
    def create_validation_error(field, value, constraint):
        exc = ValidationError(f"{field} failed: {constraint}")
        exc.field = field
        exc.value = value
        exc.constraint = constraint
        return exc

    @staticmethod
    def create_api_error(status_code, response_body):
        if status_code == 404:
            return NotFoundError(f"Resource not found: {response_body}")
        elif status_code == 429:
            return RateLimitError(f"Rate limited: {response_body}")
        elif status_code >= 500:
            return ServerError(f"Server error {status_code}: {response_body}")
        else:
            return APIError(f"API error {status_code}: {response_body}")

# Builder pattern for complex exceptions
class ExceptionBuilder:
    def __init__(self, exc_type):
        self.exc_type = exc_type
        self.message = None
        self.context = {}
        self.cause = None

    def with_message(self, message):
        self.message = message
        return self

    def with_context(self, **kwargs):
        self.context.update(kwargs)
        return self

    def with_cause(self, cause):
        self.cause = cause
        return self

    def build(self):
        exc = self.exc_type(self.message, **self.context)
        if self.cause:
            raise exc from self.cause
        return exc

# Usage
error = ExceptionBuilder(ValidationError)\
    .with_message("Invalid user data")\
    .with_context(field='email', value='invalid')\
    .build()

raise error

🔐 Serializable Exceptions

# Exceptions that can be serialized for logging/transmission
import json
from datetime import datetime

class SerializableException(Exception):
    """Exception that can be converted to JSON"""

    def __init__(self, message, error_code=None, **context):
        self.message = message
        self.error_code = error_code or self.__class__.__name__
        self.context = context
        self.timestamp = datetime.now()
        self.traceback_str = None
        super().__init__(message)

    def to_dict(self):
        """Convert to dictionary for JSON serialization"""
        return {
            'error_code': self.error_code,
            'message': self.message,
            'context': self.context,
            'timestamp': self.timestamp.isoformat(),
            'type': self.__class__.__name__
        }

    def to_json(self):
        """Convert to JSON string"""
        return json.dumps(self.to_dict())

    @staticmethod
    def from_dict(data):
        """Reconstruct from dictionary"""
        return SerializableException(
            message=data['message'],
            error_code=data['error_code'],
            **data.get('context', {})
        )

# Usage with HTTP responses
class HTTPException(SerializableException):
    def __init__(self, message, status_code=400, **context):
        self.status_code = status_code
        super().__init__(message, **context)

    def to_response(self):
        """Convert to HTTP response"""
        return {
            'status_code': self.status_code,
            'body': self.to_dict()
        }

# Usage
try:
    raise HTTPException("User not found", status_code=404, user_id=123)
except HTTPException as e:
    response = e.to_response()
    print(response)  # Can send directly to client

📊 Exception with Metrics and Monitoring

# Exceptions with built-in metrics
class MetricsException(Exception):
    """Exception that tracks its own metrics"""

    _occurrence_count = 0
    _instances = []

    def __init__(self, message, severity='ERROR'):
        self.message = message
        self.severity = severity
        self.timestamp = datetime.now()

        # Track metrics
        MetricsException._occurrence_count += 1
        MetricsException._instances.append(self)

        super().__init__(message)

    @classmethod
    def get_metrics(cls):
        """Get aggregated metrics"""
        return {
            'total_count': cls._occurrence_count,
            'by_type': self._count_by_type(),
            'by_severity': self._count_by_severity(),
            'recent_errors': cls._instances[-10:]
        }

    @staticmethod
    def _count_by_type():
        counts = {}
        for exc in MetricsException._instances:
            exc_type = type(exc).__name__
            counts[exc_type] = counts.get(exc_type, 0) + 1
        return counts

    @staticmethod
    def _count_by_severity():
        counts = {}
        for exc in MetricsException._instances:
            severity = exc.severity
            counts[severity] = counts.get(severity, 0) + 1
        return counts

class AlertableException(MetricsException):
    """Exception that triggers alerts"""

    _alert_threshold = 5
    _recent_alerts = []

    def __init__(self, message, should_alert=True):
        super().__init__(message, severity='CRITICAL')
        self.should_alert = should_alert

        if should_alert:
            AlertableException._recent_alerts.append(self)
            if len(AlertableException._recent_alerts) >= self._alert_threshold:
                self._trigger_alert()

    @staticmethod
    def _trigger_alert():
        # Send alert to monitoring system
        alert_msg = f"High error rate detected: {len(AlertableException._recent_alerts)} critical errors"
        # send_to_monitoring_system(alert_msg)
        AlertableException._recent_alerts.clear()

# Usage
try:
    raise AlertableException("Database connection failed")
except AlertableException as e:
    print(MetricsException.get_metrics())

🔍 Contextual Stack Information

# Store full stack information in exception
import traceback
import inspect

class DetailedStackException(Exception):
    """Exception with detailed stack information"""

    def __init__(self, message):
        self.message = message
        self.stack_frames = []
        self._capture_stack()
        super().__init__(message)

    def _capture_stack(self):
        """Capture detailed information about each frame"""
        for frame_info in inspect.stack()[2:]:  # Skip this method and caller
            self.stack_frames.append({
                'filename': frame_info.filename,
                'line_number': frame_info.lineno,
                'function': frame_info.function,
                'code': frame_info.code_context[0].strip() if frame_info.code_context else '',
                'locals': self._safe_repr(frame_info.frame.f_locals)
            })

    @staticmethod
    def _safe_repr(obj, max_length=100):
        """Safe representation that doesn't expose secrets"""
        try:
            result = repr(obj)
            if len(result) > max_length:
                result = result[:max_length] + "..."
            # Redact sensitive keys
            for sensitive in ['password', 'token', 'secret', 'key']:
                if sensitive in result.lower():
                    result = f"<{sensitive} redacted>"
            return result
        except:
            return "<error getting repr>"

    def get_full_report(self):
        """Get detailed error report"""
        report = f"Exception: {self.message}\n"
        report += "\nStack Trace:\n"
        for frame in self.stack_frames:
            report += f"  File {frame['filename']}, line {frame['line_number']}, in {frame['function']}\n"
            report += f"    {frame['code']}\n"
        return report

# Usage
def problematic_function():
    secret_data = "password123"
    raise DetailedStackException("Something failed")

try:
    problematic_function()
except DetailedStackException as e:
    print(e.get_full_report())

🎯 Domain-Specific Exception Hierarchies

# E-commerce domain exceptions
class ECommerceError(Exception):
    """Base exception for e-commerce"""
    pass

class OrderError(ECommerceError):
    """Order-related errors"""
    pass

class PaymentError(ECommerceError):
    """Payment-related errors"""
    pass

class InventoryError(ECommerceError):
    """Inventory-related errors"""
    pass

class OrderValidationError(OrderError):
    def __init__(self, message, order_id):
        self.order_id = order_id
        super().__init__(f"Order {order_id}: {message}")

class InsufficientInventoryError(InventoryError):
    def __init__(self, sku, requested, available):
        self.sku = sku
        self.requested = requested
        self.available = available
        super().__init__(
            f"SKU {sku}: need {requested}, have {available}"
        )

class PaymentDeclinedError(PaymentError):
    def __init__(self, reason, order_id, amount):
        self.reason = reason
        self.order_id = order_id
        self.amount = amount
        super().__init__(f"Payment declined: {reason}")

# Usage with domain logic
def process_order(order):
    try:
        validate_order(order)
        check_inventory(order)
        process_payment(order)
        create_shipment(order)
    except OrderValidationError as e:
        logger.warning(f"Invalid order {e.order_id}")
        return {'status': 'rejected', 'reason': str(e)}
    except InsufficientInventoryError as e:
        logger.warning(f"Out of stock: {e.sku}")
        return {'status': 'backorder', 'sku': e.sku, 'available': e.available}
    except PaymentDeclinedError as e:
        logger.error(f"Payment failed for order {e.order_id}: {e.reason}")
        return {'status': 'payment_failed', 'reason': e.reason}
    except ECommerceError as e:
        logger.error(f"Unexpected e-commerce error: {e}")
        return {'status': 'error', 'reason': str(e)}

    return {'status': 'success'}

🔗 Exception Middleware for Web Frameworks

# Exception handling middleware for Flask/Django
class ExceptionMiddleware:
    """Middleware for centralized exception handling"""

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

    def __call__(self, environ, start_response):
        try:
            return self.app(environ, start_response)
        except SerializableException as e:
            response = e.to_response()
            return self.send_response(start_response, response)
        except Exception as e:
            logger.exception("Unhandled exception")
            response = {
                'status_code': 500,
                'body': {'error': 'Internal server error'}
            }
            return self.send_response(start_response, response)

    @staticmethod
    def send_response(start_response, response):
        status = f"{response['status_code']} Internal Server Error"
        headers = [('Content-Type', 'application/json')]
        start_response(status, headers)
        return [json.dumps(response['body']).encode()]

# Flask example
from flask import Flask, jsonify

app = Flask(__name__)

@app.errorhandler(SerializableException)
def handle_serializable_error(error):
    response = error.to_response()
    return jsonify(response['body']), response['status_code']

@app.errorhandler(Exception)
def handle_generic_error(error):
    logger.exception("Unhandled exception")
    return jsonify({
        'error': 'Internal server error',
        'message': 'An unexpected error occurred'
    }), 500

🎯 Key Takeaways

• ✅ Use factory pattern for consistent exception creation

• ✅ Implement builder pattern for complex exceptions

• ✅ Make exceptions JSON-serializable for logging

• ✅ Track metrics and monitor exception patterns

• ✅ Capture detailed stack information safely

• ✅ Design domain-specific exception hierarchies

• ✅ Implement exception middleware for web apps