Commit Helper MCP

""" Error Handlers Provides decorators and utilities for consistent error handling across the application with logging, monitoring, and recovery mechanisms. """ import logging import functools from typing import Dict, Any, Callable, Optional, Type from .exceptions import ( CommitHelperMCPError, GitOperationError, ValidationError, ConfigurationError, RepositoryError, PluginError, ServiceError, ) from .responses import create_error_response logger = logging.getLogger(__name__) def handle_errors( default_response: Optional[Dict[str, Any]] = None, log_errors: bool = True, reraise: bool = False, ): """ Decorator for comprehensive error handling. Args: default_response: Default response to return on error log_errors: Whether to log errors reraise: Whether to reraise exceptions after handling """ def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(*args, **kwargs) -> Dict[str, Any]: try: return func(*args, **kwargs) except CommitHelperMCPError as e: if log_errors: log_error_context(e, func.__name__) response = create_error_response(e) if reraise: raise return response except Exception as e: # Convert unexpected exceptions to CommitHelperMCPError mcp_error = ServiceError( f"Unexpected error in {func.__name__}: {str(e)}", service_name=func.__module__, cause=e, ) if log_errors: log_error_context(mcp_error, func.__name__) response = create_error_response(mcp_error) if reraise: raise mcp_error from e return ( response or default_response or {"error": "An unexpected error occurred", "success": False} ) return wrapper return decorator def handle_git_errors(func: Callable) -> Callable: """Decorator specifically for git operation error handling.""" @functools.wraps(func) def wrapper(*args, **kwargs) -> Dict[str, Any]: try: return func(*args, **kwargs) except GitOperationError: # Already a git error, re-raise raise except Exception as e: # Convert to GitOperationError git_error = GitOperationError( f"Git operation failed in {func.__name__}: {str(e)}", cause=e ) raise git_error from e return wrapper def handle_validation_errors(func: Callable) -> Callable: """Decorator specifically for validation error handling.""" @functools.wraps(func) def wrapper(*args, **kwargs) -> Dict[str, Any]: try: return func(*args, **kwargs) except ValidationError: # Already a validation error, re-raise raise except Exception as e: # Convert to ValidationError validation_error = ValidationError( f"Validation failed in {func.__name__}: {str(e)}", cause=e ) raise validation_error from e return wrapper def handle_configuration_errors(func: Callable) -> Callable: """Decorator specifically for configuration error handling.""" @functools.wraps(func) def wrapper(*args, **kwargs) -> Dict[str, Any]: try: return func(*args, **kwargs) except ConfigurationError: # Already a configuration error, re-raise raise except Exception as e: # Convert to ConfigurationError config_error = ConfigurationError( f"Configuration error in {func.__name__}: {str(e)}", cause=e ) raise config_error from e return wrapper def log_error_context(error: CommitHelperMCPError, function_name: str): """Log error with full context information.""" logger.error( f"Error in {function_name}: {error}", extra={ "error_type": error.__class__.__name__, "operation": error.context.operation, "component": error.context.component, "details": error.context.details, "function": function_name, "cause": str(error.cause) if error.cause else None, }, ) def create_error_recovery_response( error: CommitHelperMCPError, additional_context: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """Create error response with recovery information.""" response = create_error_response(error) if additional_context: response.update(additional_context) # Add recovery information if error.get_suggestions(): response["suggestions"] = error.get_suggestions() if error.get_recovery_actions(): response["recovery_actions"] = error.get_recovery_actions() return response class ErrorHandler: """Context manager for error handling.""" def __init__( self, operation: str, component: str, log_errors: bool = True, reraise: bool = True, ): self.operation = operation self.component = component self.log_errors = log_errors self.reraise = reraise def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is None: return False if issubclass(exc_type, CommitHelperMCPError): if self.log_errors: log_error_context(exc_val, self.operation) return not self.reraise # Convert unexpected exceptions mcp_error = ServiceError( f"Unexpected error in {self.operation}: {str(exc_val)}", service_name=self.component, cause=exc_val, ) if self.log_errors: log_error_context(mcp_error, self.operation) if self.reraise: raise mcp_error from exc_val return True