error_handler.py•17 kB
#!/usr/bin/env python3
"""
Error Handling Utilities
Provides comprehensive error handling, exception management, and error reporting.
"""
import sys
import traceback
import functools
from typing import Any, Callable, Dict, List, Optional, Type, Union
from datetime import datetime
from dataclasses import dataclass, field
from enum import Enum
from .logging import get_logger, ContextualLogger
logger = get_logger(__name__)
class ErrorSeverity(Enum):
"""Error severity levels."""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class ErrorCategory(Enum):
"""Error categories for classification."""
VALIDATION = "validation"
AUTHENTICATION = "authentication"
AUTHORIZATION = "authorization"
NETWORK = "network"
DATABASE = "database"
FILE_SYSTEM = "file_system"
CONFIGURATION = "configuration"
ADAPTER = "adapter"
MCP_PROTOCOL = "mcp_protocol"
SYSTEM = "system"
UNKNOWN = "unknown"
@dataclass
class ErrorContext:
"""Enhanced error context information."""
error_id: str
timestamp: datetime = field(default_factory=datetime.utcnow)
category: ErrorCategory = ErrorCategory.UNKNOWN
severity: ErrorSeverity = ErrorSeverity.MEDIUM
component: Optional[str] = None
operation: Optional[str] = None
user_id: Optional[str] = None
request_id: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for logging."""
return {
"error_id": self.error_id,
"timestamp": self.timestamp.isoformat(),
"category": self.category.value,
"severity": self.severity.value,
"component": self.component,
"operation": self.operation,
"user_id": self.user_id,
"request_id": self.request_id,
"metadata": self.metadata
}
class AnyDocsError(Exception):
"""Base exception class for AnyDocs MCP."""
def __init__(
self,
message: str,
category: ErrorCategory = ErrorCategory.UNKNOWN,
severity: ErrorSeverity = ErrorSeverity.MEDIUM,
context: Optional[ErrorContext] = None,
cause: Optional[Exception] = None
):
super().__init__(message)
self.message = message
self.category = category
self.severity = severity
self.context = context or ErrorContext(
error_id=self._generate_error_id(),
category=category,
severity=severity
)
self.cause = cause
def _generate_error_id(self) -> str:
"""Generate unique error ID."""
import uuid
return str(uuid.uuid4())[:8]
def to_dict(self) -> Dict[str, Any]:
"""Convert exception to dictionary."""
return {
"error": self.__class__.__name__,
"message": self.message,
"context": self.context.to_dict() if self.context else None,
"cause": str(self.cause) if self.cause else None
}
class ValidationError(AnyDocsError):
"""Validation error exception."""
def __init__(self, message: str, field: Optional[str] = None, **kwargs):
super().__init__(message, category=ErrorCategory.VALIDATION, **kwargs)
self.field = field
class AuthenticationError(AnyDocsError):
"""Authentication error exception."""
def __init__(self, message: str = "Authentication failed", **kwargs):
super().__init__(message, category=ErrorCategory.AUTHENTICATION, severity=ErrorSeverity.HIGH, **kwargs)
class AuthorizationError(AnyDocsError):
"""Authorization error exception."""
def __init__(self, message: str = "Access denied", **kwargs):
super().__init__(message, category=ErrorCategory.AUTHORIZATION, severity=ErrorSeverity.HIGH, **kwargs)
class NetworkError(AnyDocsError):
"""Network-related error exception."""
def __init__(self, message: str, status_code: Optional[int] = None, **kwargs):
super().__init__(message, category=ErrorCategory.NETWORK, **kwargs)
self.status_code = status_code
class DatabaseError(AnyDocsError):
"""Database-related error exception."""
def __init__(self, message: str, **kwargs):
super().__init__(message, category=ErrorCategory.DATABASE, severity=ErrorSeverity.HIGH, **kwargs)
class ConfigurationError(AnyDocsError):
"""Configuration error exception."""
def __init__(self, message: str, config_key: Optional[str] = None, **kwargs):
super().__init__(message, category=ErrorCategory.CONFIGURATION, severity=ErrorSeverity.HIGH, **kwargs)
self.config_key = config_key
class AdapterError(AnyDocsError):
"""Document adapter error exception."""
def __init__(self, message: str, adapter_type: Optional[str] = None, **kwargs):
super().__init__(message, category=ErrorCategory.ADAPTER, **kwargs)
self.adapter_type = adapter_type
class MCPProtocolError(AnyDocsError):
"""MCP protocol error exception."""
def __init__(self, message: str, **kwargs):
super().__init__(message, category=ErrorCategory.MCP_PROTOCOL, severity=ErrorSeverity.HIGH, **kwargs)
class ErrorHandler:
"""Centralized error handler."""
def __init__(self, logger: Optional[ContextualLogger] = None):
"""Initialize error handler."""
self.logger = logger or get_logger("error_handler")
self.error_counts: Dict[str, int] = {}
def handle_error(
self,
error: Exception,
context: Optional[ErrorContext] = None,
reraise: bool = True
) -> ErrorContext:
"""Handle and log an error.
Args:
error: Exception to handle
context: Optional error context
reraise: Whether to reraise the exception
Returns:
Error context
"""
# Create or enhance context
if isinstance(error, AnyDocsError):
error_context = error.context or context
if context and error.context:
# Merge contexts
error_context.metadata.update(context.metadata)
if context.component and not error_context.component:
error_context.component = context.component
if context.operation and not error_context.operation:
error_context.operation = context.operation
else:
error_context = context or ErrorContext(
error_id=self._generate_error_id(),
category=self._classify_error(error),
severity=self._determine_severity(error)
)
# Log error
self._log_error(error, error_context)
# Track error count
self._track_error(error_context)
# Reraise if requested
if reraise:
if isinstance(error, AnyDocsError):
raise error
else:
# Wrap in AnyDocsError
raise AnyDocsError(
str(error),
context=error_context,
cause=error
) from error
return error_context
def _classify_error(self, error: Exception) -> ErrorCategory:
"""Classify error by type."""
error_type = type(error).__name__.lower()
if "validation" in error_type or "value" in error_type:
return ErrorCategory.VALIDATION
elif "auth" in error_type or "permission" in error_type:
return ErrorCategory.AUTHENTICATION
elif "network" in error_type or "connection" in error_type or "http" in error_type:
return ErrorCategory.NETWORK
elif "database" in error_type or "sql" in error_type:
return ErrorCategory.DATABASE
elif "file" in error_type or "io" in error_type:
return ErrorCategory.FILE_SYSTEM
elif "config" in error_type:
return ErrorCategory.CONFIGURATION
else:
return ErrorCategory.SYSTEM
def _determine_severity(self, error: Exception) -> ErrorSeverity:
"""Determine error severity."""
if isinstance(error, (KeyboardInterrupt, SystemExit)):
return ErrorSeverity.CRITICAL
elif isinstance(error, (ConnectionError, DatabaseError)):
return ErrorSeverity.HIGH
elif isinstance(error, (ValueError, TypeError)):
return ErrorSeverity.MEDIUM
else:
return ErrorSeverity.LOW
def _log_error(self, error: Exception, context: ErrorContext) -> None:
"""Log error with full context."""
log_data = {
"error_type": type(error).__name__,
"error_message": str(error),
"traceback": traceback.format_exc(),
**context.to_dict()
}
# Log based on severity
if context.severity == ErrorSeverity.CRITICAL:
self.logger.critical("Critical error occurred", **log_data)
elif context.severity == ErrorSeverity.HIGH:
self.logger.error("High severity error occurred", **log_data)
elif context.severity == ErrorSeverity.MEDIUM:
self.logger.warning("Medium severity error occurred", **log_data)
else:
self.logger.info("Low severity error occurred", **log_data)
def _track_error(self, context: ErrorContext) -> None:
"""Track error occurrence for metrics."""
key = f"{context.category.value}:{context.severity.value}"
self.error_counts[key] = self.error_counts.get(key, 0) + 1
def _generate_error_id(self) -> str:
"""Generate unique error ID."""
import uuid
return str(uuid.uuid4())[:8]
def get_error_stats(self) -> Dict[str, Any]:
"""Get error statistics."""
return {
"total_errors": sum(self.error_counts.values()),
"error_counts": self.error_counts.copy(),
"timestamp": datetime.utcnow().isoformat()
}
# Global error handler instance
_global_error_handler: Optional[ErrorHandler] = None
def get_error_handler() -> ErrorHandler:
"""Get global error handler instance."""
global _global_error_handler
if _global_error_handler is None:
_global_error_handler = ErrorHandler()
return _global_error_handler
def handle_error(
error: Exception,
context: Optional[ErrorContext] = None,
reraise: bool = True
) -> ErrorContext:
"""Handle error using global error handler."""
return get_error_handler().handle_error(error, context, reraise)
def safe_execute(
func: Callable,
*args,
default_return: Any = None,
error_context: Optional[ErrorContext] = None,
**kwargs
) -> Any:
"""Safely execute a function with error handling.
Args:
func: Function to execute
*args: Function arguments
default_return: Value to return on error
error_context: Error context
**kwargs: Function keyword arguments
Returns:
Function result or default_return on error
"""
try:
return func(*args, **kwargs)
except Exception as e:
handle_error(e, error_context, reraise=False)
return default_return
def error_handler_decorator(
category: ErrorCategory = ErrorCategory.UNKNOWN,
severity: ErrorSeverity = ErrorSeverity.MEDIUM,
reraise: bool = True,
default_return: Any = None
):
"""Decorator for automatic error handling.
Args:
category: Error category
severity: Error severity
reraise: Whether to reraise exceptions
default_return: Default return value on error
"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
context = ErrorContext(
error_id=str(uuid.uuid4())[:8],
category=category,
severity=severity,
component=func.__module__,
operation=func.__name__
)
try:
handle_error(e, context, reraise)
except Exception:
if not reraise:
return default_return
raise
return default_return
@functools.wraps(func)
async def async_wrapper(*args, **kwargs):
try:
return await func(*args, **kwargs)
except Exception as e:
context = ErrorContext(
error_id=str(uuid.uuid4())[:8],
category=category,
severity=severity,
component=func.__module__,
operation=func.__name__
)
try:
handle_error(e, context, reraise)
except Exception:
if not reraise:
return default_return
raise
return default_return
# Return appropriate wrapper based on function type
import asyncio
if asyncio.iscoroutinefunction(func):
return async_wrapper
else:
return wrapper
return decorator
# Retry mechanism with exponential backoff
def retry_on_error(
max_attempts: int = 3,
delay: float = 1.0,
backoff_factor: float = 2.0,
exceptions: tuple = (Exception,)
):
"""Decorator for retrying operations on specific exceptions.
Args:
max_attempts: Maximum number of retry attempts
delay: Initial delay between retries
backoff_factor: Multiplicative factor for delay increase
exceptions: Tuple of exceptions to retry on
"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
import time
last_exception = None
current_delay = delay
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt < max_attempts - 1:
logger.warning(
f"Operation failed, retrying in {current_delay}s",
function=func.__name__,
attempt=attempt + 1,
max_attempts=max_attempts,
error=str(e)
)
time.sleep(current_delay)
current_delay *= backoff_factor
else:
logger.error(
f"Operation failed after {max_attempts} attempts",
function=func.__name__,
error=str(e)
)
# All attempts failed, raise the last exception
raise last_exception
@functools.wraps(func)
async def async_wrapper(*args, **kwargs):
import asyncio
last_exception = None
current_delay = delay
for attempt in range(max_attempts):
try:
return await func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt < max_attempts - 1:
logger.warning(
f"Operation failed, retrying in {current_delay}s",
function=func.__name__,
attempt=attempt + 1,
max_attempts=max_attempts,
error=str(e)
)
await asyncio.sleep(current_delay)
current_delay *= backoff_factor
else:
logger.error(
f"Operation failed after {max_attempts} attempts",
function=func.__name__,
error=str(e)
)
# All attempts failed, raise the last exception
raise last_exception
# Return appropriate wrapper based on function type
import asyncio
if asyncio.iscoroutinefunction(func):
return async_wrapper
else:
return wrapper
return decorator