Base Error Classes

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

ErrorSeverity

Categorizes errors by their severity level for logging, monitoring, and user feedback.

---

ErrorSeverity

def ErrorSeverity(
    args:VAR_POSITIONAL, kwds:VAR_KEYWORD
):

Error severity levels for categorization and handling.

ErrorContext

A structured container for error context metadata. This allows errors to carry rich contextual information without cluttering the error message itself.

---

ErrorContext

def ErrorContext(
    job_id:Optional=None, plugin_id:Optional=None, worker_pid:Optional=None, session_id:Optional=None,
    user_id:Optional=None, operation:Optional=None, timestamp:str=<factory>, extra:Dict=<factory>
)->None:

Structured context information for errors.

Example: ErrorContext

# Create context with common fields
ctx = ErrorContext(
    job_id="job-123",
    plugin_id="whisper_large",
    worker_pid=12345,
    operation="transcribe_audio"
)

print("Context:", ctx)
print("\nAs dict:", ctx.to_dict())

Context: ErrorContext(job_id='job-123', plugin_id='whisper_large', worker_pid=12345, session_id=None, user_id=None, operation='transcribe_audio', timestamp='2025-10-28T14:53:10.872694', extra={})

As dict: {'job_id': 'job-123', 'plugin_id': 'whisper_large', 'worker_pid': 12345, 'operation': 'transcribe_audio', 'timestamp': '2025-10-28T14:53:10.872694', 'extra': {}}

# Create context with extra domain-specific fields
ctx2 = ErrorContext(
    job_id="job-456",
    operation="validate_resources",
    extra={
        "gpu_id": 0,
        "memory_required_mb": 8192,
        "resource_type": "GPU"
    }
)

print("Context with extra:", ctx2.to_dict())

Context with extra: {'job_id': 'job-456', 'operation': 'validate_resources', 'timestamp': '2025-10-28T14:53:10.889187', 'extra': {'gpu_id': 0, 'memory_required_mb': 8192, 'resource_type': 'GPU'}}

# Round-trip serialization
ctx_dict = ctx2.to_dict()
ctx_restored = ErrorContext.from_dict(ctx_dict)
print("Restored:", ctx_restored)
print("\nMatches original:", ctx_restored.to_dict() == ctx_dict)

Restored: ErrorContext(job_id='job-456', plugin_id=None, worker_pid=None, session_id=None, user_id=None, operation='validate_resources', timestamp='2025-10-28T14:53:10.889187', extra={'gpu_id': 0, 'memory_required_mb': 8192, 'resource_type': 'GPU'})

Matches original: True

BaseError

The foundation exception class providing: - Dual messaging: User-friendly message + optional debug details - Context propagation: Rich contextual metadata via ErrorContext - Serialization: to_dict() and from_dict() for crossing process boundaries - Error chaining: Preserves original exception via cause - Severity levels: Categorization for logging and monitoring - Retry hints: Indicates if error is transient and retryable

---

BaseError

def BaseError(
    message:str, # User-friendly error message
    debug_info:Optional=None, # Optional developer details
    context:Optional=None, # Structured error context
    severity:ErrorSeverity=<ErrorSeverity.ERROR: 'error'>, # Error severity level
    is_retryable:bool=False, # Whether error is transient/retryable
    cause:Optional=None, # Original exception if chaining
):

Base exception class with rich context and dual messaging.

Example: Basic Usage

# Simple error with just a message
error1 = BaseError("Configuration file not found")
print("User message:", error1.get_user_message())
print("Debug message:", error1.get_debug_message())
print("Severity:", error1.severity.value)

User message: Configuration file not found
Debug message: Configuration file not found
Severity: error

# Error with debug information
error2 = BaseError(
    message="Failed to save settings",
    debug_info="JSONDecodeError at line 42: Unexpected token '}'",
    severity=ErrorSeverity.ERROR
)
print("User message:", error2.get_user_message())
print("Debug message:", error2.get_debug_message())

User message: Failed to save settings
Debug message: Failed to save settings | Debug: JSONDecodeError at line 42: Unexpected token '}'

Example: With Context

# Error with rich context
ctx = ErrorContext(
    job_id="job-789",
    plugin_id="whisper_large",
    operation="load_model",
    extra={"model_path": "/models/whisper-large-v3.pt", "gpu_id": 0}
)

error3 = BaseError(
    message="GPU out of memory",
    debug_info="CUDA error: out of memory (tried to allocate 8.50 GiB)",
    context=ctx,
    severity=ErrorSeverity.CRITICAL,
    is_retryable=False
)

print("Error:", error3)
print("\nContext:", error3.context.to_dict())
print("\nRetryable:", error3.is_retryable)

Error: GPU out of memory

Context: {'job_id': 'job-789', 'plugin_id': 'whisper_large', 'operation': 'load_model', 'timestamp': '2025-10-28T14:53:10.966470', 'extra': {'model_path': '/models/whisper-large-v3.pt', 'gpu_id': 0}}

Retryable: False

Example: Error Chaining

# Simulate catching and re-raising with context
try:
    # Simulate an original error
    raise FileNotFoundError("/models/config.json")
except FileNotFoundError as e:
    # Wrap with our error type
    error4 = BaseError(
        message="Plugin configuration not found",
        debug_info="Check that the model was downloaded correctly",
        context=ErrorContext(plugin_id="whisper_base", operation="load_config"),
        cause=e
    )
    
    print("User message:", error4.get_user_message())
    print("Debug message:", error4.get_debug_message())
    print("\nOriginal cause:", error4.cause)

User message: Plugin configuration not found
Debug message: Plugin configuration not found | Debug: Check that the model was downloaded correctly | Caused by: FileNotFoundError: /models/config.json

Original cause: /models/config.json

Example: Serialization (for Worker Processes)

import json

# Create an error
error5 = BaseError(
    message="Job failed",
    debug_info="Worker process terminated unexpectedly",
    context=ErrorContext(
        job_id="job-999",
        worker_pid=54321,
        operation="transcribe"
    ),
    severity=ErrorSeverity.ERROR,
    is_retryable=True
)

# Serialize to dict
error_dict = error5.to_dict()
print("Serialized:")
print(json.dumps(error_dict, indent=2))

Serialized:
{
  "error_type": "BaseError",
  "message": "Job failed",
  "severity": "error",
  "is_retryable": true,
  "context": {
    "job_id": "job-999",
    "worker_pid": 54321,
    "operation": "transcribe",
    "timestamp": "2026-04-14T05:31:46.062317",
    "extra": {}
  },
  "debug_info": "Worker process terminated unexpectedly"
}

# Deserialize back to error
error_restored = BaseError.from_dict(error_dict)
print("\nRestored error:")
print(f"  Message: {error_restored.message}")
print(f"  Debug: {error_restored.debug_info}")
print(f"  Severity: {error_restored.severity.value}")
print(f"  Retryable: {error_restored.is_retryable}")
print(f"  Context: {error_restored.context.to_dict()}")

Restored error:
  Message: Job failed
  Debug: Worker process terminated unexpectedly
  Severity: error
  Retryable: True
  Context: {'job_id': 'job-999', 'worker_pid': 54321, 'operation': 'transcribe', 'timestamp': '2025-10-28T14:53:10.997481', 'extra': {}}

Example: Usage in Try/Except

def risky_operation(should_fail=True):
    """Simulated operation that may fail."""
    if should_fail:
        raise BaseError(
            message="Operation failed",
            debug_info="Insufficient permissions to write to /var/log",
            context=ErrorContext(operation="write_logs"),
            is_retryable=False
        )
    return "Success!"

# Catching the error
try:
    result = risky_operation(should_fail=True)
except BaseError as e:
    print(f"Caught error: {e.get_user_message()}")
    print(f"Can retry: {e.is_retryable}")
    print(f"Severity: {e.severity.value}")
    
    # In a real app, you might:
    # - Log the debug message: logger.error(e.get_debug_message())
    # - Show user the user message: display_alert(e.get_user_message())
    # - Send to monitoring: send_to_sentry(e.to_dict())

Caught error: Operation failed
Can retry: False
Severity: error