Metadata-Version: 2.4
Name: cjm-error-handling
Version: 0.0.1
Summary: Structured error handling with context propagation, user-friendly messaging, and serialization support for reusable Python libraries.
Home-page: https://github.com/cj-mills/cjm-error-handling
Author: Christian J. Mills
Author-email: 9126128+cj-mills@users.noreply.github.com
License: Apache Software License 2.0
Keywords: nbdev jupyter notebook python
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Natural Language :: English
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: License :: OSI Approved :: Apache Software License
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: dev
Dynamic: author
Dynamic: author-email
Dynamic: classifier
Dynamic: description
Dynamic: description-content-type
Dynamic: home-page
Dynamic: keywords
Dynamic: license
Dynamic: license-file
Dynamic: provides-extra
Dynamic: requires-python
Dynamic: summary

# cjm-error-handling


<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->

## Install

``` sh
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

``` mermaid
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

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

#### Classes

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

``` python
@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."
```

``` python
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

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

#### Classes

``` python
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
```

``` python
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
```

``` python
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
```

``` python
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
```

``` python
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

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

#### Functions

```` python
@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)
        ```
    """
````

```` python
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
        ```
    """
````

```` python
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
        ```
    """
````

```` python
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"
            )
        ```
    """
````
