cjm-error-handling 0.0.2

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_errors
    utils_helpers --> core_base

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."
    
    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."
    
    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."""
            result = {
                'error_type': self.__class__.__name__,
        "Serialize error to dictionary for transmission across process boundaries."
    
    def from_dict(cls, data: Dict[str, Any]) -> 'BaseError':  # Reconstructed error
            """Reconstruct error from dictionary representation."""
            # 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."

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,                              # 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,                # Validation errors typically need fixes, not retries
        cause: Optional[Exception] = None,         # Original exception if chaining
        validation_errors: Optional[Dict[str, Any]] = None  # Structured validation details
    )
    "Raised when validation fails."
    
    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,                # Validation errors typically need fixes, not retries
            cause: Optional[Exception] = None,         # Original exception if chaining
            validation_errors: Optional[Dict[str, Any]] = None  # Structured validation details
        )
        "Initialize validation error with optional structured validation details."
    
    def to_dict(self) -> Dict[str, Any]:  # Dictionary representation
            """Serialize error including validation details."""
            result = super().to_dict()
            if self.validation_errors
        "Serialize error including validation details."

class ConfigurationError:
    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,                # Config issues usually need manual fixes
        cause: Optional[Exception] = None,         # Original exception if chaining
        config_path: Optional[str] = None          # Path to problematic config file
    )
    "Raised when configuration operations fail."
    
    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,                # Config issues usually need manual fixes
            cause: Optional[Exception] = None,         # Original exception if chaining
            config_path: Optional[str] = None          # Path to problematic config file
        )
        "Initialize configuration error with optional config file path."
    
    def to_dict(self) -> Dict[str, Any]:  # Dictionary representation
            """Serialize error including config path."""
            result = super().to_dict()
            if self.config_path
        "Serialize error including config path."

class ResourceError:
    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.WARNING,  # Often transient
        is_retryable: bool = True,                 # Resource conflicts may be temporary
        cause: Optional[Exception] = None,         # Original exception if chaining
        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."
    
    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.WARNING,  # Often transient
            is_retryable: bool = True,                 # Resource conflicts may be temporary
            cause: Optional[Exception] = None,         # Original exception if chaining
            resource_type: Optional[str] = None,       # "GPU", "Memory", "Disk", etc.
            suggested_action: Optional[str] = None     # Guidance for resolution
        )
        "Initialize resource error with resource type and suggested action."
    
    def to_dict(self) -> Dict[str, Any]:  # Dictionary representation
            """Serialize error including resource type and suggested action."""
            result = super().to_dict()
            if self.resource_type
        "Serialize error including resource type and suggested action."

class PluginError:
    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,                # Plugin errors usually need fixes
        cause: Optional[Exception] = None,         # Original exception if chaining
        plugin_id: Optional[str] = None,           # ID of problematic plugin
        plugin_name: Optional[str] = None          # Name of problematic plugin
    )
    "Raised when plugin operations fail."
    
    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,                # Plugin errors usually need fixes
            cause: Optional[Exception] = None,         # Original exception if chaining
            plugin_id: Optional[str] = None,           # ID of problematic plugin
            plugin_name: Optional[str] = None          # Name of problematic plugin
        )
        "Initialize plugin error with plugin ID and name."
    
    def to_dict(self) -> Dict[str, Any]:  # Dictionary representation
            """Serialize error including plugin ID and name."""
            result = super().to_dict()
            if self.plugin_id
        "Serialize error including plugin ID and name."

class WorkerError:
    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 = True,                 # Worker errors may be transient
        cause: Optional[Exception] = None,         # Original exception if chaining
        worker_type: Optional[str] = None,         # "transcription", "llm", etc.
        job_id: Optional[str] = None               # Job that failed
    )
    "Raised when worker process operations fail."
    
    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 = True,                 # Worker errors may be transient
            cause: Optional[Exception] = None,         # Original exception if chaining
            worker_type: Optional[str] = None,         # "transcription", "llm", etc.
            job_id: Optional[str] = None               # Job that failed
        )
        "Initialize worker error with worker type and job ID."
    
    def to_dict(self) -> Dict[str, Any]:  # Dictionary representation
            """Serialize error including worker type and job ID."""
            result = super().to_dict()
            if self.worker_type
        "Serialize error including worker type and job ID."

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."

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, e.g. {'job_id': 'job_id'}
)
    "Decorator that wraps function errors with structured error handling."

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."

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."