Helper Utilities

Utilities for easy adoption and incremental migration to structured error handling

Adoption Utilities

These utilities make it easy to incrementally adopt structured error handling in existing codebases without requiring large-scale refactoring.

error_boundary

A context manager that automatically enriches exceptions with context and converts them to structured errors.

---

error_boundary

def error_boundary(
    error_type:Type=BaseError, # Error type to raise
    message:Optional=None, # User-facing message (uses original if None)
    context:Optional=None, # Error context
    operation:Optional=None, # Operation name (added to context)
    job_id:Optional=None, # Job ID (added to context)
    plugin_id:Optional=None, # Plugin ID (added to context)
    worker_pid:Optional=None, # Worker PID (added to context)
    severity:ErrorSeverity=<ErrorSeverity.ERROR: 'error'>, # Error severity
    is_retryable:Optional=None, # Override retryable flag
    catch:tuple=(<class 'Exception'>,), # Exception types to catch
    extra_context:VAR_KEYWORD
):

Context manager that catches exceptions and wraps them in structured errors.

Example: error_boundary

# Basic usage - wrap existing code
try:
    with error_boundary(
        error_type=ConfigurationError,
        operation="load_config",
        message="Failed to load configuration"
    ):
        # Simulate config loading error
        raise FileNotFoundError("/config/settings.json")
except ConfigurationError as e:
    print(f"Caught: {type(e).__name__}")
    print(f"Message: {e.get_user_message()}")
    print(f"Debug: {e.get_debug_message()}")
    print(f"Context: {e.context.operation}")

Caught: ConfigurationError
Message: Failed to load configuration
Debug: Failed to load configuration | Debug: Original error: FileNotFoundError: /config/settings.json | Caused by: FileNotFoundError: /config/settings.json
Context: load_config

# With context fields
try:
    with error_boundary(
        error_type=PluginError,
        operation="initialize_plugin",
        plugin_id="whisper_large",
        model_path="/models/whisper-large-v3.pt"  # Extra context
    ):
        raise ImportError("No module named 'torch'")
except PluginError as e:
    print(f"\n{type(e).__name__}: {e.get_user_message()}")
    print(f"Plugin: {e.context.plugin_id}")
    print(f"Extra: {e.context.extra}")

PluginError: No module named 'torch'
Plugin: whisper_large
Extra: {'model_path': '/models/whisper-large-v3.pt'}

with_error_handling

A decorator that automatically wraps function errors with context.

---

with_error_handling

def with_error_handling(
    error_type:Type=BaseError, # Error type to raise
    message:Optional=None, # User-facing message
    operation:Optional=None, # Operation name (uses function name if None)
    severity:ErrorSeverity=<ErrorSeverity.ERROR: 'error'>, # Error severity
    is_retryable:Optional=None, # Override retryable flag
    catch:tuple=(<class 'Exception'>,), # Exception types to catch
    context_from_args:Optional=None, # Map arg names to context fields, e.g. {'job_id': 'job_id'}
):

Decorator that wraps function errors with structured error handling.

Example: with_error_handling decorator

# Simple decorator usage
@with_error_handling(
    error_type=ValidationError,
    message="Configuration validation failed"
)
def validate_config(config):
    """Validate a configuration dictionary."""
    if 'model_id' not in config:
        raise ValueError("model_id is required")
    return True

# Test it
try:
    validate_config({})  # Missing model_id
except ValidationError as e:
    print(f"Caught: {type(e).__name__}")
    print(f"Message: {e.get_user_message()}")
    print(f"Operation: {e.context.operation}")

Caught: ValidationError
Message: Configuration validation failed
Operation: validate_config

# With context from arguments
@with_error_handling(
    error_type=WorkerError,
    context_from_args={'job_id': 'job_id', 'worker_pid': 'worker_pid'}
)
def execute_job(job_id, worker_pid, data):
    """Execute a job in a worker."""
    if data is None:
        raise RuntimeError("No data provided")
    return f"Processed {data}"

# Test it
try:
    execute_job(job_id="job-123", worker_pid=54321, data=None)
except WorkerError as e:
    print(f"\n{type(e).__name__}: {e.get_user_message()}")
    print(f"Job ID: {e.context.job_id}")
    print(f"Worker PID: {e.context.worker_pid}")
    print(f"Operation: {e.context.operation}")

WorkerError: No data provided
Job ID: job-123
Worker PID: 54321
Operation: execute_job

wrap_exception

Helper function to wrap an existing exception in a structured error type.

---

wrap_exception

def wrap_exception(
    exception:Exception, # Original exception to wrap
    error_type:Type=BaseError, # Error type to create
    message:Optional=None, # User-facing message (uses exception if None)
    context:Optional=None, # Error context
    severity:ErrorSeverity=<ErrorSeverity.ERROR: 'error'>, # Error severity
    is_retryable:Optional=None, # Override retryable flag
    context_kwargs:VAR_KEYWORD
)->BaseError: # Wrapped structured error

Wrap an existing exception in a structured error type.

Example: wrap_exception

# Wrap a caught exception
try:
    import json
    json.loads('{bad json}')
except json.JSONDecodeError as e:
    # Wrap in our error type
    error = wrap_exception(
        e,
        error_type=ConfigurationError,
        message="Failed to parse configuration file",
        operation="load_config",
        plugin_id="whisper_base"
    )
    
    print(f"Wrapped error: {type(error).__name__}")
    print(f"Message: {error.get_user_message()}")
    print(f"Debug: {error.get_debug_message()}")
    print(f"Context: {error.context.to_dict()}")

Wrapped error: ConfigurationError
Message: Failed to parse configuration file
Debug: Failed to parse configuration file | Debug: Original: JSONDecodeError: Expecting property name enclosed in double quotes: line 1 column 2 (char 1) | Caused by: JSONDecodeError: Expecting property name enclosed in double quotes: line 1 column 2 (char 1)
Context: {'plugin_id': 'whisper_base', 'operation': 'load_config', 'timestamp': '2026-04-14T05:31:46.024619', 'extra': {}}

chain_error

Helper for explicit error chaining with enriched context.

---

chain_error

def chain_error(
    base_error:BaseError, # Original structured error
    new_message:str, # New user-facing message
    error_type:Optional=None, # New error type (uses base type if None)
    additional_context:Optional=None, # Additional context to add
    operation:Optional=None, # Update operation name
    severity:Optional=None, # Override severity
)->BaseError: # New error with chained context

Chain a structured error with additional context.

Example: chain_error

# Simulate layered error handling
def load_config_file():
    """Low-level config loading."""
    raise ConfigurationError(
        message="Config file not found",
        debug_info="/config/whisper.json does not exist",
        context=ErrorContext(operation="read_file")
    )

def initialize_plugin():
    """Mid-level plugin initialization."""
    try:
        load_config_file()
    except ConfigurationError as e:
        # Chain with plugin context
        raise chain_error(
            e,
            new_message="Plugin initialization failed",
            error_type=PluginError,
            operation="initialize_plugin",
            additional_context={"plugin_name": "Whisper Large"}
        )

# Test chained errors
try:
    initialize_plugin()
except PluginError as e:
    print(f"Error type: {type(e).__name__}")
    print(f"Message: {e.get_user_message()}")
    print(f"Operation: {e.context.operation}")
    print(f"Extra context: {e.context.extra}")
    print(f"\nDebug (includes chain):")
    print(f"  {e.get_debug_message()}")

Error type: PluginError
Message: Plugin initialization failed
Operation: initialize_plugin
Extra context: {'plugin_name': 'Whisper Large'}

Debug (includes chain):
  Plugin initialization failed | Debug: Config file not found | Debug: /config/whisper.json does not exist | Caused by: ConfigurationError: Config file not found

Usage Summary

These utilities enable different adoption strategies:

1. Context Manager (Quickest)

Wrap existing code blocks:

with error_boundary(error_type=PluginError, plugin_id="whisper"):
    existing_code()

2. Decorator (Cleanest)

Annotate functions:

@with_error_handling(error_type=WorkerError)
def process_job(job_id):
    ...

3. Explicit Wrapping (Most Control)

Manually wrap in try/except:

try:
    risky_operation()
except Exception as e:
    error = wrap_exception(e, error_type=PluginError, plugin_id="whisper")
    raise error

4. Error Chaining (For Layers)

Add context as errors propagate:

try:
    lower_level_function()
except ConfigurationError as e:
    raise chain_error(e, "Higher level failed", operation="top_level")