Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |