File size: 8,677 Bytes
8be45a9 c087535 8be45a9 c087535 8be45a9 c087535 8be45a9 65e7d4d 6ff8c59 8be45a9 c087535 8be45a9 c087535 8be45a9 c087535 8be45a9 6ff8c59 8be45a9 6ff8c59 8be45a9 6ff8c59 8be45a9 c087535 8be45a9 c087535 8be45a9 c087535 8be45a9 c087535 65e7d4d 8be45a9 8e4018d 8be45a9 8e4018d 8be45a9 65e7d4d afc51ae 8be45a9 afc51ae 8be45a9 afc51ae 8be45a9 |
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 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 |
"""
Error handling utilities for the MONA application
Fixed version with proper exception handling and logging
"""
import functools
import traceback
from typing import Any, Callable, Dict, Optional, Union
from utils.logging import get_logger, log_error, log_warning
class DataError(Exception):
"""Custom exception for data-related errors"""
def __init__(self, message: str, error_code: str = None, details: Dict = None):
self.message = message
self.error_code = error_code or "DATA_ERROR"
self.details = details or {}
super().__init__(self.message)
class ValidationError(Exception):
"""Custom exception for validation errors"""
def __init__(self, message: str, field: str = None, value: Any = None):
self.message = message
self.field = field
self.value = value
super().__init__(self.message)
class ProcessingError(Exception):
"""Custom exception for processing errors"""
def __init__(self, message: str, step: str = None, context: Dict = None):
self.message = message
self.step = step
self.context = context or {}
super().__init__(self.message)
def handle_data_exceptions(func: Callable) -> Callable:
"""
Decorator to handle data-related exceptions
Args:
func: Function to wrap with exception handling
Returns:
Wrapped function with exception handling
"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except DataError as e:
log_error(f"Data error in {func.__name__}: {e.message}",
error=e,
extra_data={"error_code": e.error_code, "details": e.details})
raise
except ValidationError as e:
log_error(f"Validation error in {func.__name__}: {e.message}",
error=e,
extra_data={"field": e.field, "value": e.value})
raise
except ProcessingError as e:
log_error(f"Processing error in {func.__name__}: {e.message}",
error=e,
extra_data={"step": e.step, "context": e.context})
raise
except Exception as e:
log_error(f"Unexpected error in {func.__name__}: {str(e)}",
error=e)
raise ProcessingError(f"Unexpected error in {func.__name__}",
context={"original_error": str(e)})
return wrapper
def safe_execute(func: Callable, *args, default_return=None, **kwargs) -> tuple:
"""
Safely execute a function and return success status with result
Args:
func: Function to execute
*args: Positional arguments for the function
default_return: Default value to return on error
**kwargs: Keyword arguments for the function
Returns:
tuple: (success: bool, result: Any, error: Optional[Exception])
"""
try:
result = func(*args, **kwargs)
return True, result, None
except Exception as e:
log_error(f"Error executing {func.__name__}: {str(e)}", error=e)
return False, default_return, e
def validate_data(data: Any, validation_rules: Dict) -> bool:
"""
Validate data against a set of rules
Args:
data: Data to validate
validation_rules: Dictionary of validation rules
Returns:
bool: True if validation passes
Raises:
ValidationError: If validation fails
"""
if not isinstance(validation_rules, dict):
raise ValidationError("Validation rules must be a dictionary")
for field, rules in validation_rules.items():
if isinstance(data, dict):
value = data.get(field)
else:
value = getattr(data, field, None)
# Check required fields
if rules.get('required', False) and value is None:
raise ValidationError(f"Field '{field}' is required", field=field, value=value)
# Check data type
if 'type' in rules and value is not None:
expected_type = rules['type']
if not isinstance(value, expected_type):
raise ValidationError(
f"Field '{field}' must be of type {expected_type.__name__}, got {type(value).__name__}",
field=field,
value=value
)
# Check value range for numbers
if 'min_value' in rules and isinstance(value, (int, float)):
if value < rules['min_value']:
raise ValidationError(
f"Field '{field}' must be >= {rules['min_value']}, got {value}",
field=field,
value=value
)
if 'max_value' in rules and isinstance(value, (int, float)):
if value > rules['max_value']:
raise ValidationError(
f"Field '{field}' must be <= {rules['max_value']}, got {value}",
field=field,
value=value
)
# Check string length
if 'min_length' in rules and isinstance(value, str):
if len(value) < rules['min_length']:
raise ValidationError(
f"Field '{field}' must be at least {rules['min_length']} characters",
field=field,
value=value
)
if 'max_length' in rules and isinstance(value, str):
if len(value) > rules['max_length']:
raise ValidationError(
f"Field '{field}' must be at most {rules['max_length']} characters",
field=field,
value=value
)
return True
def format_error_response(error: Exception) -> Dict:
"""
Format an error into a standardized response
Args:
error: Exception to format
Returns:
Dict: Formatted error response
"""
if isinstance(error, (DataError, ValidationError, ProcessingError)):
return {
"success": False,
"error_type": error.__class__.__name__,
"message": error.message,
"details": getattr(error, 'details', {}),
"field": getattr(error, 'field', None),
"error_code": getattr(error, 'error_code', None)
}
else:
return {
"success": False,
"error_type": "UnexpectedError",
"message": str(error),
"details": {}
}
def retry_on_failure(max_retries: int = 3, delay: float = 1.0, exceptions: tuple = (Exception,)):
"""
Decorator to retry function execution on failure
Args:
max_retries: Maximum number of retry attempts
delay: Delay between retries in seconds
exceptions: Tuple of exceptions to catch and retry on
Returns:
Decorator function
"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
import time
last_exception = None
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt < max_retries:
log_warning(f"Attempt {attempt + 1} failed for {func.__name__}: {str(e)}. Retrying in {delay}s...")
time.sleep(delay)
else:
log_error(f"All {max_retries + 1} attempts failed for {func.__name__}", error=e)
raise last_exception
return wrapper
return decorator
# Context manager for error handling
class ErrorHandler:
"""Context manager for comprehensive error handling"""
def __init__(self, operation_name: str, raise_on_error: bool = True):
self.operation_name = operation_name
self.raise_on_error = raise_on_error
self.error = None
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is not None:
self.error = exc_val
log_error(f"Error in {self.operation_name}: {str(exc_val)}", error=exc_val)
if not self.raise_on_error:
return True # Suppress the exception
return False # Let the exception propagate |