# 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_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
``` 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."
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."
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
``` python
from cjm_error_handling.core.errors import (
ValidationError,
ConfigurationError,
ResourceError,
PluginError,
WorkerError
)
```
#### Classes
``` python
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."
```
``` python
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."
```
``` python
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."
```
``` python
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."
```
``` python
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
``` 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."
```
``` 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, e.g. {'job_id': 'job_id'}
)
"Decorator that wraps function errors with structured error handling."
```
``` 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."
```
``` 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."
```
Raw data
{
"_id": null,
"home_page": "https://github.com/cj-mills/cjm-error-handling",
"name": "cjm-error-handling",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": null,
"keywords": "nbdev jupyter notebook python",
"author": "Christian J. Mills",
"author_email": "9126128+cj-mills@users.noreply.github.com",
"download_url": "https://files.pythonhosted.org/packages/b4/b9/dce3b754aff60648b3f1a9d89a079bdf9d00d19cdc4c7d743df5c917659b/cjm_error_handling-0.0.2.tar.gz",
"platform": null,
"description": "# cjm-error-handling\n\n\n<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->\n\n## Install\n\n``` sh\npip install cjm_error_handling\n```\n\n## Project Structure\n\n nbs/\n \u251c\u2500\u2500 core/ (2)\n \u2502 \u251c\u2500\u2500 base.ipynb # Foundation classes for structured error handling with context propagation and dual messaging\n \u2502 \u2514\u2500\u2500 errors.ipynb # Concrete error classes for common failure scenarios in library ecosystems\n \u2514\u2500\u2500 utils/ (1)\n \u2514\u2500\u2500 helpers.ipynb # Utilities for easy adoption and incremental migration to structured error handling\n\nTotal: 3 notebooks across 2 directories\n\n## Module Dependencies\n\n``` mermaid\ngraph LR\n core_base[core.base<br/>Base Error Classes]\n core_errors[core.errors<br/>Domain-Specific Error Types]\n utils_helpers[utils.helpers<br/>Helper Utilities]\n\n core_errors --> core_base\n utils_helpers --> core_errors\n utils_helpers --> core_base\n```\n\n*3 cross-module dependencies detected*\n\n## CLI Reference\n\nNo CLI commands found in this project.\n\n## Module Overview\n\nDetailed documentation for each module in the project:\n\n### Base Error Classes (`base.ipynb`)\n\n> Foundation classes for structured error handling with context\n> propagation and dual messaging\n\n#### Import\n\n``` python\nfrom cjm_error_handling.core.base import (\n ErrorSeverity,\n ErrorContext,\n BaseError\n)\n```\n\n#### Classes\n\n``` python\nclass ErrorSeverity(Enum):\n \"Error severity levels for categorization and handling.\"\n```\n\n``` python\n@dataclass\nclass ErrorContext:\n \"Structured context information for errors.\"\n \n job_id: Optional[str] # Job identifier\n plugin_id: Optional[str] # Plugin identifier\n worker_pid: Optional[int] # Worker process ID\n session_id: Optional[str] # Session identifier\n user_id: Optional[str] # User identifier\n operation: Optional[str] # Operation being performed\n timestamp: str = field(...) # When error occurred\n extra: Dict[str, Any] = field(...) # Domain-specific context\n \n def to_dict(self) -> Dict[str, Any]: # Dictionary representation\n \"\"\"Convert context to dictionary for serialization.\"\"\"\n result = asdict(self)\n # Remove None values for cleaner serialization\n return {k: v for k, v in result.items() if v is not None}\n \"Convert context to dictionary for serialization.\"\n \n def from_dict(cls, data: Dict[str, Any]) -> 'ErrorContext': # ErrorContext instance\n \"\"\"Create ErrorContext from dictionary.\"\"\"\n # Extract known fields\n known_fields = {'job_id', 'plugin_id', 'worker_pid', 'session_id',\n 'user_id', 'operation', 'timestamp'}\n kwargs = {k: v for k, v in data.items() if k in known_fields}\n \"Create ErrorContext from dictionary.\"\n```\n\n``` python\nclass BaseError:\n def __init__(\n self,\n message: str, # User-friendly error message\n debug_info: Optional[str] = None, # Optional developer details\n context: Optional[ErrorContext] = None, # Structured error context\n severity: ErrorSeverity = ErrorSeverity.ERROR, # Error severity level\n is_retryable: bool = False, # Whether error is transient/retryable\n cause: Optional[Exception] = None # Original exception if chaining\n )\n \"Base exception class with rich context and dual messaging.\"\n \n def __init__(\n self,\n message: str, # User-friendly error message\n debug_info: Optional[str] = None, # Optional developer details\n context: Optional[ErrorContext] = None, # Structured error context\n severity: ErrorSeverity = ErrorSeverity.ERROR, # Error severity level\n is_retryable: bool = False, # Whether error is transient/retryable\n cause: Optional[Exception] = None # Original exception if chaining\n )\n \"Initialize error with message, context, and metadata.\"\n \n def get_user_message(self) -> str: # User-friendly message\n \"\"\"Get the user-friendly error message.\"\"\"\n return self.message\n \n def get_debug_message(self) -> str: # Debug message with details\n \"Get the user-friendly error message.\"\n \n def get_debug_message(self) -> str: # Debug message with details\n \"\"\"Get detailed debug information.\"\"\"\n parts = [self.message]\n \n if self.debug_info\n \"Get detailed debug information.\"\n \n def to_dict(self) -> Dict[str, Any]: # Dictionary representation\n \"\"\"Serialize error to dictionary for transmission across process boundaries.\"\"\"\n result = {\n 'error_type': self.__class__.__name__,\n \"Serialize error to dictionary for transmission across process boundaries.\"\n \n def from_dict(cls, data: Dict[str, Any]) -> 'BaseError': # Reconstructed error\n \"\"\"Reconstruct error from dictionary representation.\"\"\"\n # Reconstruct context\n context = ErrorContext.from_dict(data.get('context', {}))\n \n # Reconstruct severity\n severity_str = data.get('severity', 'error')\n severity = ErrorSeverity(severity_str)\n \n # Note: We can't reconstruct the original cause exception,\n \"Reconstruct error from dictionary representation.\"\n```\n\n### Domain-Specific Error Types (`errors.ipynb`)\n\n> Concrete error classes for common failure scenarios in library\n> ecosystems\n\n#### Import\n\n``` python\nfrom cjm_error_handling.core.errors import (\n ValidationError,\n ConfigurationError,\n ResourceError,\n PluginError,\n WorkerError\n)\n```\n\n#### Classes\n\n``` python\nclass ValidationError:\n def __init__(\n self,\n message: str, # User-friendly error message\n debug_info: Optional[str] = None, # Optional developer details\n context: Optional[ErrorContext] = None, # Structured error context\n severity: ErrorSeverity = ErrorSeverity.ERROR, # Error severity level\n is_retryable: bool = False, # Validation errors typically need fixes, not retries\n cause: Optional[Exception] = None, # Original exception if chaining\n validation_errors: Optional[Dict[str, Any]] = None # Structured validation details\n )\n \"Raised when validation fails.\"\n \n def __init__(\n self,\n message: str, # User-friendly error message\n debug_info: Optional[str] = None, # Optional developer details\n context: Optional[ErrorContext] = None, # Structured error context\n severity: ErrorSeverity = ErrorSeverity.ERROR, # Error severity level\n is_retryable: bool = False, # Validation errors typically need fixes, not retries\n cause: Optional[Exception] = None, # Original exception if chaining\n validation_errors: Optional[Dict[str, Any]] = None # Structured validation details\n )\n \"Initialize validation error with optional structured validation details.\"\n \n def to_dict(self) -> Dict[str, Any]: # Dictionary representation\n \"\"\"Serialize error including validation details.\"\"\"\n result = super().to_dict()\n if self.validation_errors\n \"Serialize error including validation details.\"\n```\n\n``` python\nclass ConfigurationError:\n def __init__(\n self,\n message: str, # User-friendly error message\n debug_info: Optional[str] = None, # Optional developer details\n context: Optional[ErrorContext] = None, # Structured error context\n severity: ErrorSeverity = ErrorSeverity.ERROR, # Error severity level\n is_retryable: bool = False, # Config issues usually need manual fixes\n cause: Optional[Exception] = None, # Original exception if chaining\n config_path: Optional[str] = None # Path to problematic config file\n )\n \"Raised when configuration operations fail.\"\n \n def __init__(\n self,\n message: str, # User-friendly error message\n debug_info: Optional[str] = None, # Optional developer details\n context: Optional[ErrorContext] = None, # Structured error context\n severity: ErrorSeverity = ErrorSeverity.ERROR, # Error severity level\n is_retryable: bool = False, # Config issues usually need manual fixes\n cause: Optional[Exception] = None, # Original exception if chaining\n config_path: Optional[str] = None # Path to problematic config file\n )\n \"Initialize configuration error with optional config file path.\"\n \n def to_dict(self) -> Dict[str, Any]: # Dictionary representation\n \"\"\"Serialize error including config path.\"\"\"\n result = super().to_dict()\n if self.config_path\n \"Serialize error including config path.\"\n```\n\n``` python\nclass ResourceError:\n def __init__(\n self,\n message: str, # User-friendly error message\n debug_info: Optional[str] = None, # Optional developer details\n context: Optional[ErrorContext] = None, # Structured error context\n severity: ErrorSeverity = ErrorSeverity.WARNING, # Often transient\n is_retryable: bool = True, # Resource conflicts may be temporary\n cause: Optional[Exception] = None, # Original exception if chaining\n resource_type: Optional[str] = None, # \"GPU\", \"Memory\", \"Disk\", etc.\n suggested_action: Optional[str] = None # Guidance for resolution\n )\n \"Raised when resource conflicts or unavailability prevent operation.\"\n \n def __init__(\n self,\n message: str, # User-friendly error message\n debug_info: Optional[str] = None, # Optional developer details\n context: Optional[ErrorContext] = None, # Structured error context\n severity: ErrorSeverity = ErrorSeverity.WARNING, # Often transient\n is_retryable: bool = True, # Resource conflicts may be temporary\n cause: Optional[Exception] = None, # Original exception if chaining\n resource_type: Optional[str] = None, # \"GPU\", \"Memory\", \"Disk\", etc.\n suggested_action: Optional[str] = None # Guidance for resolution\n )\n \"Initialize resource error with resource type and suggested action.\"\n \n def to_dict(self) -> Dict[str, Any]: # Dictionary representation\n \"\"\"Serialize error including resource type and suggested action.\"\"\"\n result = super().to_dict()\n if self.resource_type\n \"Serialize error including resource type and suggested action.\"\n```\n\n``` python\nclass PluginError:\n def __init__(\n self,\n message: str, # User-friendly error message\n debug_info: Optional[str] = None, # Optional developer details\n context: Optional[ErrorContext] = None, # Structured error context\n severity: ErrorSeverity = ErrorSeverity.ERROR, # Error severity level\n is_retryable: bool = False, # Plugin errors usually need fixes\n cause: Optional[Exception] = None, # Original exception if chaining\n plugin_id: Optional[str] = None, # ID of problematic plugin\n plugin_name: Optional[str] = None # Name of problematic plugin\n )\n \"Raised when plugin operations fail.\"\n \n def __init__(\n self,\n message: str, # User-friendly error message\n debug_info: Optional[str] = None, # Optional developer details\n context: Optional[ErrorContext] = None, # Structured error context\n severity: ErrorSeverity = ErrorSeverity.ERROR, # Error severity level\n is_retryable: bool = False, # Plugin errors usually need fixes\n cause: Optional[Exception] = None, # Original exception if chaining\n plugin_id: Optional[str] = None, # ID of problematic plugin\n plugin_name: Optional[str] = None # Name of problematic plugin\n )\n \"Initialize plugin error with plugin ID and name.\"\n \n def to_dict(self) -> Dict[str, Any]: # Dictionary representation\n \"\"\"Serialize error including plugin ID and name.\"\"\"\n result = super().to_dict()\n if self.plugin_id\n \"Serialize error including plugin ID and name.\"\n```\n\n``` python\nclass WorkerError:\n def __init__(\n self,\n message: str, # User-friendly error message\n debug_info: Optional[str] = None, # Optional developer details\n context: Optional[ErrorContext] = None, # Structured error context\n severity: ErrorSeverity = ErrorSeverity.ERROR, # Error severity level\n is_retryable: bool = True, # Worker errors may be transient\n cause: Optional[Exception] = None, # Original exception if chaining\n worker_type: Optional[str] = None, # \"transcription\", \"llm\", etc.\n job_id: Optional[str] = None # Job that failed\n )\n \"Raised when worker process operations fail.\"\n \n def __init__(\n self,\n message: str, # User-friendly error message\n debug_info: Optional[str] = None, # Optional developer details\n context: Optional[ErrorContext] = None, # Structured error context\n severity: ErrorSeverity = ErrorSeverity.ERROR, # Error severity level\n is_retryable: bool = True, # Worker errors may be transient\n cause: Optional[Exception] = None, # Original exception if chaining\n worker_type: Optional[str] = None, # \"transcription\", \"llm\", etc.\n job_id: Optional[str] = None # Job that failed\n )\n \"Initialize worker error with worker type and job ID.\"\n \n def to_dict(self) -> Dict[str, Any]: # Dictionary representation\n \"\"\"Serialize error including worker type and job ID.\"\"\"\n result = super().to_dict()\n if self.worker_type\n \"Serialize error including worker type and job ID.\"\n```\n\n### Helper Utilities (`helpers.ipynb`)\n\n> Utilities for easy adoption and incremental migration to structured\n> error handling\n\n#### Import\n\n``` python\nfrom cjm_error_handling.utils.helpers import (\n error_boundary,\n with_error_handling,\n wrap_exception,\n chain_error\n)\n```\n\n#### Functions\n\n``` python\n@contextmanager\ndef error_boundary(\n error_type: Type[BaseError] = BaseError, # Error type to raise\n message: Optional[str] = None, # User-facing message (uses original if None)\n context: Optional[ErrorContext] = None, # Error context\n operation: Optional[str] = None, # Operation name (added to context)\n job_id: Optional[str] = None, # Job ID (added to context)\n plugin_id: Optional[str] = None, # Plugin ID (added to context)\n worker_pid: Optional[int] = None, # Worker PID (added to context)\n severity: ErrorSeverity = ErrorSeverity.ERROR, # Error severity\n is_retryable: Optional[bool] = None, # Override retryable flag\n catch: tuple = (Exception,), # Exception types to catch\n **extra_context # Additional context fields\n)\n \"Context manager that catches exceptions and wraps them in structured errors.\"\n```\n\n``` python\ndef with_error_handling(\n error_type: Type[BaseError] = BaseError, # Error type to raise\n message: Optional[str] = None, # User-facing message\n operation: Optional[str] = None, # Operation name (uses function name if None)\n severity: ErrorSeverity = ErrorSeverity.ERROR, # Error severity\n is_retryable: Optional[bool] = None, # Override retryable flag\n catch: tuple = (Exception,), # Exception types to catch\n context_from_args: Optional[Dict[str, str]] = None # Map arg names to context fields, e.g. {'job_id': 'job_id'}\n)\n \"Decorator that wraps function errors with structured error handling.\"\n```\n\n``` python\ndef wrap_exception(\n exception: Exception, # Original exception to wrap\n error_type: Type[BaseError] = BaseError, # Error type to create\n message: Optional[str] = None, # User-facing message (uses exception if None)\n context: Optional[ErrorContext] = None, # Error context\n severity: ErrorSeverity = ErrorSeverity.ERROR, # Error severity\n is_retryable: Optional[bool] = None, # Override retryable flag\n **context_kwargs # Additional context fields\n) -> BaseError: # Wrapped structured error\n \"Wrap an existing exception in a structured error type.\"\n```\n\n``` python\ndef chain_error(\n base_error: BaseError, # Original structured error\n new_message: str, # New user-facing message\n error_type: Optional[Type[BaseError]] = None, # New error type (uses base type if None)\n additional_context: Optional[Dict[str, Any]] = None, # Additional context to add\n operation: Optional[str] = None, # Update operation name\n severity: Optional[ErrorSeverity] = None # Override severity\n) -> BaseError: # New error with chained context\n \"Chain a structured error with additional context.\"\n```\n",
"bugtrack_url": null,
"license": "Apache Software License 2.0",
"summary": "Structured error handling with context propagation, user-friendly messaging, and serialization support for reusable Python libraries.",
"version": "0.0.2",
"project_urls": {
"Homepage": "https://github.com/cj-mills/cjm-error-handling"
},
"split_keywords": [
"nbdev",
"jupyter",
"notebook",
"python"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "a8883b16545515b6962ba6a59a26df0d0d419945cf84d6cee2f5d1cac41e0697",
"md5": "16e82ece2c53632672d07934362ad171",
"sha256": "419525155e47b5c2b4755341ae389cc48fe5d614b0d2bdfcc2e081d50514978a"
},
"downloads": -1,
"filename": "cjm_error_handling-0.0.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "16e82ece2c53632672d07934362ad171",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 16309,
"upload_time": "2025-10-29T17:20:10",
"upload_time_iso_8601": "2025-10-29T17:20:10.164143Z",
"url": "https://files.pythonhosted.org/packages/a8/88/3b16545515b6962ba6a59a26df0d0d419945cf84d6cee2f5d1cac41e0697/cjm_error_handling-0.0.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "b4b9dce3b754aff60648b3f1a9d89a079bdf9d00d19cdc4c7d743df5c917659b",
"md5": "3eeb9aa8c272bd7ca45ea584fb2c162a",
"sha256": "2d1181a5b2f8e857594ab735441c1aba7e6f70dca48361e7d0fcc0d7f9548b81"
},
"downloads": -1,
"filename": "cjm_error_handling-0.0.2.tar.gz",
"has_sig": false,
"md5_digest": "3eeb9aa8c272bd7ca45ea584fb2c162a",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 17246,
"upload_time": "2025-10-29T17:20:11",
"upload_time_iso_8601": "2025-10-29T17:20:11.666121Z",
"url": "https://files.pythonhosted.org/packages/b4/b9/dce3b754aff60648b3f1a9d89a079bdf9d00d19cdc4c7d743df5c917659b/cjm_error_handling-0.0.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-29 17:20:11",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "cj-mills",
"github_project": "cjm-error-handling",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "cjm-error-handling"
}
JSON
