cjm-error-handling 0.0.1

Structured error handling with context propagation, user-friendly messaging, and serialization support for reusable Python libraries.

cjm-error-handling

Install

pip install cjm_error_handling

Project Structure

nbs/
├── core/ (2)
│   ├── base.ipynb    # Foundation classes for structured error handling with context propagation and dual messaging
│   └── errors.ipynb  # Concrete error classes for common failure scenarios in library ecosystems
└── utils/ (1)
    └── helpers.ipynb  # Utilities for easy adoption and incremental migration to structured error handling

Total: 3 notebooks across 2 directories

Module Dependencies

graph LR
    core_base[core.base<br/>Base Error Classes]
    core_errors[core.errors<br/>Domain-Specific Error Types]
    utils_helpers[utils.helpers<br/>Helper Utilities]

    core_errors --> core_base
    utils_helpers --> core_base
    utils_helpers --> core_errors

3 cross-module dependencies detected

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

Base Error Classes (base.ipynb)

Foundation classes for structured error handling with context propagation and dual messaging

Import

from cjm_error_handling.core.base import (
    ErrorSeverity,
    ErrorContext,
    BaseError
)

Classes

class ErrorSeverity(Enum):
    "Error severity levels for categorization and handling."

@dataclass
class ErrorContext:
    """
    Structured context information for errors.
    
    This dataclass provides a standard way to attach contextual metadata
    to errors. Use specific fields for common context (job_id, plugin_id, etc.)
    and the `extra` dict for domain-specific information.
    """
    
    job_id: Optional[str]  # Job identifier
    plugin_id: Optional[str]  # Plugin identifier
    worker_pid: Optional[int]  # Worker process ID
    session_id: Optional[str]  # Session identifier
    user_id: Optional[str]  # User identifier
    operation: Optional[str]  # Operation being performed
    timestamp: str = field(...)  # When error occurred
    extra: Dict[str, Any] = field(...)  # Domain-specific context
    
    def to_dict(self) -> Dict[str, Any]:  # Dictionary representation
            """Convert context to dictionary for serialization."""
            result = asdict(self)
            # Remove None values for cleaner serialization
            return {k: v for k, v in result.items() if v is not None}
        "Convert context to dictionary for serialization."
    
    def from_dict(cls, data: Dict[str, Any]) -> 'ErrorContext':  # ErrorContext instance
            """Create ErrorContext from dictionary."""
            # Extract known fields
            known_fields = {'job_id', 'plugin_id', 'worker_pid', 'session_id', 
                           'user_id', 'operation', 'timestamp'}
            kwargs = {k: v for k, v in data.items() if k in known_fields}
        "Create ErrorContext from dictionary."

class BaseError:
    def __init__(
        self,
        message: str,                              # User-friendly error message
        debug_info: Optional[str] = None,          # Optional developer details
        context: Optional[ErrorContext] = None,    # Structured error context
        severity: ErrorSeverity = ErrorSeverity.ERROR,  # Error severity level
        is_retryable: bool = False,                # Whether error is transient/retryable
        cause: Optional[Exception] = None          # Original exception if chaining
    )
    """
    Base exception class with rich context and dual messaging.
    
    This class provides a standardized way to create errors with:
    - User-friendly messages for end users
    - Debug information for developers
    - Structured context for logging and monitoring
    - Serialization support for crossing process boundaries
    
    All custom errors in the library ecosystem should inherit from this.
    """
    
    def __init__(
            self,
            message: str,                              # User-friendly error message
            debug_info: Optional[str] = None,          # Optional developer details
            context: Optional[ErrorContext] = None,    # Structured error context
            severity: ErrorSeverity = ErrorSeverity.ERROR,  # Error severity level
            is_retryable: bool = False,                # Whether error is transient/retryable
            cause: Optional[Exception] = None          # Original exception if chaining
        )
        "Initialize error with message, context, and metadata."
    
    def get_user_message(self) -> str:  # User-friendly message
            """Get the user-friendly error message."""
            return self.message
        
        def get_debug_message(self) -> str:  # Debug message with details
        "Get the user-friendly error message."
    
    def get_debug_message(self) -> str:  # Debug message with details
            """Get detailed debug information."""
            parts = [self.message]
            
            if self.debug_info
        "Get detailed debug information."
    
    def to_dict(self) -> Dict[str, Any]:  # Dictionary representation
            """
            Serialize error to dictionary for transmission across process boundaries.
            
            Returns a dictionary that can be JSON-serialized and used to reconstruct
            the error on the other side.
            """
            result = {
                'error_type': self.__class__.__name__,
        "Serialize error to dictionary for transmission across process boundaries.

Returns a dictionary that can be JSON-serialized and used to reconstruct
the error on the other side."
    
    def from_dict(cls, data: Dict[str, Any]) -> 'BaseError':  # Reconstructed error
            """
            Reconstruct error from dictionary representation.
            
            This is useful when receiving errors from worker processes or
            other boundaries where the original exception can't be passed directly.
            """
            # Reconstruct context
            context = ErrorContext.from_dict(data.get('context', {}))
            
            # Reconstruct severity
            severity_str = data.get('severity', 'error')
            severity = ErrorSeverity(severity_str)
            
            # Note: We can't reconstruct the original cause exception,
        "Reconstruct error from dictionary representation.

This is useful when receiving errors from worker processes or
other boundaries where the original exception can't be passed directly."

Domain-Specific Error Types (errors.ipynb)

Concrete error classes for common failure scenarios in library ecosystems

Import

from cjm_error_handling.core.errors import (
    ValidationError,
    ConfigurationError,
    ResourceError,
    PluginError,
    WorkerError
)

Classes

class ValidationError:
    def __init__(
        self,
        message: str,
        debug_info: Optional[str] = None,
        context: Optional[ErrorContext] = None,
        severity: ErrorSeverity = ErrorSeverity.ERROR,
        is_retryable: bool = False,  # Validation errors typically need fixes, not retries
        cause: Optional[Exception] = None,
        validation_errors: Optional[Dict[str, Any]] = None  # Structured validation details
    )
    """
    Raised when validation fails.
    
    Use for:
    - JSON schema validation failures
    - Resource availability validation (GPU busy, insufficient memory)
    - Input constraint violations
    - Configuration value validation
    
    Examples:
        - Plugin config doesn't match schema
        - GPU memory requirements exceed available resources
        - Invalid parameter values
    """
    
    def __init__(
            self,
            message: str,
            debug_info: Optional[str] = None,
            context: Optional[ErrorContext] = None,
            severity: ErrorSeverity = ErrorSeverity.ERROR,
            is_retryable: bool = False,  # Validation errors typically need fixes, not retries
            cause: Optional[Exception] = None,
            validation_errors: Optional[Dict[str, Any]] = None  # Structured validation details
        )
    
    def to_dict(self) -> Dict[str, Any]:
            result = super().to_dict()
            if self.validation_errors

class ConfigurationError:
    def __init__(
        self,
        message: str,
        debug_info: Optional[str] = None,
        context: Optional[ErrorContext] = None,
        severity: ErrorSeverity = ErrorSeverity.ERROR,
        is_retryable: bool = False,  # Config issues usually need manual fixes
        cause: Optional[Exception] = None,
        config_path: Optional[str] = None  # Path to problematic config file
    )
    """
    Raised when configuration operations fail.
    
    Use for:
    - Missing configuration files
    - Configuration file parse errors
    - Failed to save configuration
    - Invalid configuration structure
    
    Examples:
        - config.json not found
        - Malformed JSON in config file
        - Permission denied writing config
    """
    
    def __init__(
            self,
            message: str,
            debug_info: Optional[str] = None,
            context: Optional[ErrorContext] = None,
            severity: ErrorSeverity = ErrorSeverity.ERROR,
            is_retryable: bool = False,  # Config issues usually need manual fixes
            cause: Optional[Exception] = None,
            config_path: Optional[str] = None  # Path to problematic config file
        )
    
    def to_dict(self) -> Dict[str, Any]:
            result = super().to_dict()
            if self.config_path

class ResourceError:
    def __init__(
        self,
        message: str,
        debug_info: Optional[str] = None,
        context: Optional[ErrorContext] = None,
        severity: ErrorSeverity = ErrorSeverity.WARNING,  # Often transient
        is_retryable: bool = True,  # Resource conflicts may be temporary
        cause: Optional[Exception] = None,
        resource_type: Optional[str] = None,  # "GPU", "Memory", "Disk", etc.
        suggested_action: Optional[str] = None  # Guidance for resolution
    )
    """
    Raised when resource conflicts or unavailability prevent operation.
    
    Use for:
    - GPU busy or unavailable
    - Insufficient system memory
    - Resource locked by another process
    - Quota exceeded
    
    Examples:
        - GPU in use by another worker
        - Out of memory error
        - External process using GPU
    """
    
    def __init__(
            self,
            message: str,
            debug_info: Optional[str] = None,
            context: Optional[ErrorContext] = None,
            severity: ErrorSeverity = ErrorSeverity.WARNING,  # Often transient
            is_retryable: bool = True,  # Resource conflicts may be temporary
            cause: Optional[Exception] = None,
            resource_type: Optional[str] = None,  # "GPU", "Memory", "Disk", etc.
            suggested_action: Optional[str] = None  # Guidance for resolution
        )
    
    def to_dict(self) -> Dict[str, Any]:
            result = super().to_dict()
            if self.resource_type

class PluginError:
    def __init__(
        self,
        message: str,
        debug_info: Optional[str] = None,
        context: Optional[ErrorContext] = None,
        severity: ErrorSeverity = ErrorSeverity.ERROR,
        is_retryable: bool = False,  # Plugin errors usually need fixes
        cause: Optional[Exception] = None,
        plugin_id: Optional[str] = None,  # ID of problematic plugin
        plugin_name: Optional[str] = None  # Name of problematic plugin
    )
    """
    Raised when plugin operations fail.
    
    Use for:
    - Plugin not found or unavailable
    - Plugin initialization failures
    - Plugin execution errors
    - Missing plugin dependencies
    
    Examples:
        - Plugin entry point not found
        - Plugin initialization failed
        - Plugin execute() raised an error
        - Required dependencies not installed
    """
    
    def __init__(
            self,
            message: str,
            debug_info: Optional[str] = None,
            context: Optional[ErrorContext] = None,
            severity: ErrorSeverity = ErrorSeverity.ERROR,
            is_retryable: bool = False,  # Plugin errors usually need fixes
            cause: Optional[Exception] = None,
            plugin_id: Optional[str] = None,  # ID of problematic plugin
            plugin_name: Optional[str] = None  # Name of problematic plugin
        )
    
    def to_dict(self) -> Dict[str, Any]:
            result = super().to_dict()
            if self.plugin_id

class WorkerError:
    def __init__(
        self,
        message: str,
        debug_info: Optional[str] = None,
        context: Optional[ErrorContext] = None,
        severity: ErrorSeverity = ErrorSeverity.ERROR,
        is_retryable: bool = True,  # Worker errors may be transient
        cause: Optional[Exception] = None,
        worker_type: Optional[str] = None,  # "transcription", "llm", etc.
        job_id: Optional[str] = None  # Job that failed
    )
    """
    Raised when worker process operations fail.
    
    Use for:
    - Worker process crashes
    - Worker communication failures
    - Job execution failures
    - Worker timeout errors
    
    Examples:
        - Worker process terminated unexpectedly
        - Queue communication timeout
        - Job failed in worker
        - Worker failed to start
    """
    
    def __init__(
            self,
            message: str,
            debug_info: Optional[str] = None,
            context: Optional[ErrorContext] = None,
            severity: ErrorSeverity = ErrorSeverity.ERROR,
            is_retryable: bool = True,  # Worker errors may be transient
            cause: Optional[Exception] = None,
            worker_type: Optional[str] = None,  # "transcription", "llm", etc.
            job_id: Optional[str] = None  # Job that failed
        )
    
    def to_dict(self) -> Dict[str, Any]:
            result = super().to_dict()
            if self.worker_type

Helper Utilities (helpers.ipynb)

Utilities for easy adoption and incremental migration to structured error handling

Import

from cjm_error_handling.utils.helpers import (
    error_boundary,
    with_error_handling,
    wrap_exception,
    chain_error
)

Functions

@contextmanager
def error_boundary(
    error_type: Type[BaseError] = BaseError,  # Error type to raise
    message: Optional[str] = None,  # User-facing message (uses original if None)
    context: Optional[ErrorContext] = None,  # Error context
    operation: Optional[str] = None,  # Operation name (added to context)
    job_id: Optional[str] = None,  # Job ID (added to context)
    plugin_id: Optional[str] = None,  # Plugin ID (added to context)
    worker_pid: Optional[int] = None,  # Worker PID (added to context)
    severity: ErrorSeverity = ErrorSeverity.ERROR,  # Error severity
    is_retryable: Optional[bool] = None,  # Override retryable flag
    catch: tuple = (Exception,),  # Exception types to catch
    **extra_context  # Additional context fields
)
    """
    Context manager that catches exceptions and wraps them in structured errors.
    
    This allows you to wrap existing code with minimal changes and get
    automatic error enrichment with context.
    
    Example:
        ```python
        with error_boundary(
            error_type=PluginError,
            operation="load_plugin",
            plugin_id="whisper_large"
        ):
            # Existing code that might fail
            plugin.initialize(config)
        ```
    """

def with_error_handling(
    error_type: Type[BaseError] = BaseError,  # Error type to raise
    message: Optional[str] = None,  # User-facing message
    operation: Optional[str] = None,  # Operation name (uses function name if None)
    severity: ErrorSeverity = ErrorSeverity.ERROR,  # Error severity
    is_retryable: Optional[bool] = None,  # Override retryable flag
    catch: tuple = (Exception,),  # Exception types to catch
    context_from_args: Optional[Dict[str, str]] = None  # Map arg names to context fields
)
    """
    Decorator that wraps function errors with structured error handling.
    
    This allows you to add error handling to functions without modifying
    their implementation.
    
    Args:
        error_type: Error class to raise
        message: User-facing error message
        operation: Operation name (defaults to function name)
        severity: Error severity level
        is_retryable: Whether error is retryable
        catch: Tuple of exception types to catch
        context_from_args: Map function arg names to context fields.
            Example: {'job_id': 'job_id', 'plugin': 'plugin_id'}
    
    Example:
        ```python
        @with_error_handling(
            error_type=WorkerError,
            operation="execute_job",
            context_from_args={'job_id': 'job_id'}
        )
        def run_job(job_id, data):
            # Function implementation
            pass
        ```
    """

def wrap_exception(
    exception: Exception,  # Original exception to wrap
    error_type: Type[BaseError] = BaseError,  # Error type to create
    message: Optional[str] = None,  # User-facing message (uses exception if None)
    context: Optional[ErrorContext] = None,  # Error context
    severity: ErrorSeverity = ErrorSeverity.ERROR,  # Error severity
    is_retryable: Optional[bool] = None,  # Override retryable flag
    **context_kwargs  # Additional context fields
) -> BaseError:  # Wrapped structured error
    """
    Wrap an existing exception in a structured error type.
    
    This is useful when you catch an exception and want to convert it
    to a structured error before re-raising or returning.
    
    Example:
        ```python
        try:
            load_model()
        except Exception as e:
            error = wrap_exception(
                e,
                error_type=PluginError,
                message="Failed to load model",
                plugin_id="whisper_large"
            )
            raise error
        ```
    """

def chain_error(
    base_error: BaseError,  # Original structured error
    new_message: str,  # New user-facing message
    error_type: Optional[Type[BaseError]] = None,  # New error type (uses base type if None)
    additional_context: Optional[Dict[str, Any]] = None,  # Additional context to add
    operation: Optional[str] = None,  # Update operation name
    severity: Optional[ErrorSeverity] = None  # Override severity
) -> BaseError:  # New error with chained context
    """
    Chain a structured error with additional context.
    
    This is useful when an error propagates up through layers and you want
    to add context at each layer.
    
    Example:
        ```python
        try:
            load_plugin_config()
        except ConfigurationError as e:
            # Add higher-level context
            raise chain_error(
                e,
                new_message="Failed to initialize plugin",
                error_type=PluginError,
                operation="plugin_init"
            )
        ```
    """