""" Robust Error Handling Utilities This module provides utilities for safely handling ErrorResponse objects and preventing common unpacking errors. """ import logging from typing import Any, Tuple, Union, Optional from .security import ErrorResponse logger = logging.getLogger(__name__) def extract_error_message(error_response: ErrorResponse) -> str: """ Safely extract error message from ErrorResponse object. Args: error_response: ErrorResponse object Returns: Error message string """ try: # Try different possible attribute names for error message for attr in ['message', 'error_message', 'details', 'description']: if hasattr(error_response, attr): value = getattr(error_response, attr) if value: return str(value) # Fallback to string representation return str(error_response) except Exception as e: logger.error(f"Error extracting message from ErrorResponse: {e}") return "Unknown error occurred" def safe_unpack_result(result: Any, expected_format: str = "tuple") -> Tuple[bool, Any, Optional[str]]: """ Safely unpack results that might be ErrorResponse objects or other types. Args: result: The result to unpack expected_format: Expected format ("tuple", "single", "dict") Returns: Tuple of (success, unpacked_result, error_message) """ try: # Check if result is an ErrorResponse object if isinstance(result, ErrorResponse): error_msg = extract_error_message(result) logger.warning(f"Received ErrorResponse: {error_msg}") return False, None, error_msg # Handle different expected formats if expected_format == "tuple": if isinstance(result, (tuple, list)) and len(result) == 2: return True, result, None else: error_msg = f"Expected tuple of length 2, got {type(result)} with length {len(result) if hasattr(result, '__len__') else 'N/A'}" return False, None, error_msg elif expected_format == "single": return True, result, None elif expected_format == "dict": if isinstance(result, dict): return True, result, None else: error_msg = f"Expected dict, got {type(result)}" return False, None, error_msg else: # Unknown format, try to return as-is return True, result, None except Exception as e: error_msg = f"Error during safe unpacking: {str(e)}" logger.error(error_msg) return False, None, error_msg def handle_message_processing_result(result: Any) -> Tuple[str, Any]: """ Handle results from message processing functions that might return ErrorResponse objects or tuples. Args: result: Result from message processing function Returns: Tuple of (response_message, context) """ # Check if result is an ErrorResponse first if is_error_response(result): response = extract_error_message(result) context = {} return response, context # Check if result is already a tuple/list if isinstance(result, (tuple, list)) and len(result) == 2: response, context = result return response, context # Check if result is a string (most common case for RealAPIConversationalTravelChat) if isinstance(result, str): return result, {} # Try to unpack if it might be a coroutine or other object success, unpacked_result, error_msg = safe_unpack_result(result, "tuple") if not success: # Handle error case response = f"Error processing message: {error_msg}" context = {} # Default empty context return response, context # Success case - check if unpacked result is a tuple if isinstance(unpacked_result, (tuple, list)) and len(unpacked_result) == 2: response, context = unpacked_result return response, context else: # Fallback if unpacked_result is not a tuple response = str(unpacked_result) if unpacked_result is not None else "No response" context = {} return response, context def is_error_response(obj: Any) -> bool: """ Check if an object is an ErrorResponse. Args: obj: Object to check Returns: True if object is ErrorResponse, False otherwise """ return isinstance(obj, ErrorResponse) or ( hasattr(obj, 'error_type') and hasattr(obj, 'error_code') ) def create_safe_response(result: Any, default_context: Any = None) -> Tuple[str, Any]: """ Create a safe response from any result type. Args: result: Result from any function default_context: Default context to use if result is not a tuple Returns: Tuple of (response, context) """ if is_error_response(result): response = extract_error_message(result) context = default_context or {} return response, context elif isinstance(result, (tuple, list)) and len(result) == 2: return result[0], result[1] elif isinstance(result, str): return result, default_context or {} else: response = f"Unexpected result type: {type(result)}" context = default_context or {} return response, context