# Error handling for integration layer with fallback mechanisms import logging import traceback from dataclasses import dataclass from enum import Enum from typing import Any, Callable, Dict, Optional, TypeVar logger = logging.getLogger(__name__) T = TypeVar("T") class ErrorType(Enum): """Types of errors that can occur during query processing.""" DATABASE_CONNECTION = "database_connection" DATABASE_QUERY = "database_query" SQL_VALIDATION = "sql_validation" SQL_INJECTION = "sql_injection" DOCUMENT_RETRIEVAL = "document_retrieval" LLM_GENERATION = "llm_generation" INTENT_CLASSIFICATION = "intent_classification" AGENT_EXECUTION = "agent_execution" TIMEOUT = "timeout" UNKNOWN = "unknown" @dataclass class ErrorContext: """Context information for error handling.""" error_type: ErrorType message: str original_error: Optional[Exception] = None query: Optional[str] = None additional_info: Optional[Dict[str, Any]] = None def to_dict(self) -> Dict[str, Any]: """Convert error context to dictionary.""" return { "error_type": self.error_type.value, "message": self.message, "query": self.query, "additional_info": self.additional_info, "original_error": str(self.original_error) if self.original_error else None, } class ErrorHandler: """Handles errors with fallback mechanisms and user-friendly messages.""" # Turkish user-friendly error messages ERROR_MESSAGES = { ErrorType.DATABASE_CONNECTION: "Veritabanı bağlantısı kurulamadı. Lütfen daha sonra tekrar deneyin.", ErrorType.DATABASE_QUERY: "Veritabanı sorgusu başarısız oldu. Lütfen sorunuzu farklı şekilde sormayı deneyin.", ErrorType.SQL_VALIDATION: "Oluşturulan sorgu geçersiz. Lütfen sorunuzu daha açık ifade edin.", ErrorType.SQL_INJECTION: "Güvenlik nedeniyle bu sorgu işlenemedi.", ErrorType.DOCUMENT_RETRIEVAL: "Dokümanlar aranamadı. Lütfen daha sonra tekrar deneyin.", ErrorType.LLM_GENERATION: "Yanıt oluşturulamadı. Lütfen daha sonra tekrar deneyin.", ErrorType.INTENT_CLASSIFICATION: "Sorgunuz anlaşılamadı. Lütfen farklı kelimeler kullanarak tekrar deneyin.", ErrorType.AGENT_EXECUTION: "İşlem tamamlanamadı. Lütfen daha sonra tekrar deneyin.", ErrorType.TIMEOUT: "İşlem zaman aşımına uğradı. Lütfen daha kısa bir sorgu deneyin.", ErrorType.UNKNOWN: "Beklenmeyen bir hata oluştu. Lütfen daha sonra tekrar deneyin.", } def __init__(self, enable_fallback: bool = True): """Initialize error handler with fallback configuration.""" self._enable_fallback = enable_fallback self._error_count: Dict[ErrorType, int] = {et: 0 for et in ErrorType} self._fallback_handlers: Dict[ErrorType, Callable[..., Any]] = {} @property def enable_fallback(self) -> bool: """Check if fallback is enabled.""" return self._enable_fallback @enable_fallback.setter def enable_fallback(self, value: bool) -> None: """Set fallback enabled state.""" self._enable_fallback = value @property def error_counts(self) -> Dict[ErrorType, int]: """Get error counts by type.""" return self._error_count.copy() def reset_counts(self) -> None: """Reset all error counts.""" self._error_count = {et: 0 for et in ErrorType} def register_fallback( self, error_type: ErrorType, handler: Callable[..., Any], ) -> None: """Register a fallback handler for specific error type.""" self._fallback_handlers[error_type] = handler def handle( self, error: Exception, error_type: Optional[ErrorType] = None, query: Optional[str] = None, fallback_value: Optional[T] = None, ) -> ErrorContext: """Handle an error and return context with user-friendly message.""" detected_type = error_type or self._detect_error_type(error) self._error_count[detected_type] += 1 context = ErrorContext( error_type=detected_type, message=self.ERROR_MESSAGES.get(detected_type, self.ERROR_MESSAGES[ErrorType.UNKNOWN]), original_error=error, query=query, additional_info={"traceback": traceback.format_exc()}, ) logger.error( f"Error handled: type={detected_type.value}, query={query}, error={str(error)}", exc_info=True, ) return context def execute_with_fallback( self, func: Callable[..., T], error_type: ErrorType, fallback_value: T, query: Optional[str] = None, *args: Any, **kwargs: Any, ) -> T: """Execute function with fallback on error.""" try: return func(*args, **kwargs) except Exception as e: self.handle(e, error_type, query) if self._enable_fallback and error_type in self._fallback_handlers: try: return self._fallback_handlers[error_type](query, e, *args, **kwargs) except Exception: pass return fallback_value def _detect_error_type(self, error: Exception) -> ErrorType: """Detect error type from exception.""" error_str = str(error).lower() error_type_name = type(error).__name__.lower() if "connection" in error_str or "connect" in error_type_name: return ErrorType.DATABASE_CONNECTION elif "timeout" in error_str or "timed out" in error_str: return ErrorType.TIMEOUT elif "injection" in error_str or "forbidden" in error_str: return ErrorType.SQL_INJECTION elif "sql" in error_str or "query" in error_str: return ErrorType.DATABASE_QUERY elif "validation" in error_str or "invalid" in error_str: return ErrorType.SQL_VALIDATION elif "retriev" in error_str or "document" in error_str: return ErrorType.DOCUMENT_RETRIEVAL elif "llm" in error_str or "generat" in error_str: return ErrorType.LLM_GENERATION elif "intent" in error_str or "classif" in error_str: return ErrorType.INTENT_CLASSIFICATION elif "agent" in error_str or "execution" in error_str: return ErrorType.AGENT_EXECUTION return ErrorType.UNKNOWN def get_user_message(self, error_type: ErrorType) -> str: """Get user-friendly message for error type.""" return self.ERROR_MESSAGES.get(error_type, self.ERROR_MESSAGES[ErrorType.UNKNOWN]) def get_most_frequent_error(self) -> Optional[ErrorType]: """Get the most frequently occurring error type.""" if all(count == 0 for count in self._error_count.values()): return None return max(self._error_count, key=lambda k: self._error_count[k]) _error_handler_instance: Optional[ErrorHandler] = None def get_error_handler(enable_fallback: bool = True) -> ErrorHandler: """Get or create singleton error handler instance.""" global _error_handler_instance if _error_handler_instance is None: _error_handler_instance = ErrorHandler(enable_fallback=enable_fallback) return _error_handler_instance def reset_error_handler() -> None: """Reset the singleton error handler instance.""" global _error_handler_instance _error_handler_instance = None def handle_error( error: Exception, error_type: Optional[ErrorType] = None, query: Optional[str] = None, ) -> ErrorContext: """Convenience function to handle errors using singleton handler.""" handler = get_error_handler() return handler.handle(error, error_type, query)