Spaces:
Sleeping
Sleeping
File size: 5,601 Bytes
3f9f85b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 | """
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
|