|
""" |
|
Error handling utilities for the application |
|
""" |
|
import logging |
|
import traceback |
|
from functools import wraps |
|
from typing import Any, Callable, Optional, Union |
|
import streamlit as st |
|
|
|
|
|
class DataError(Exception): |
|
"""Custom exception for data-related errors""" |
|
pass |
|
|
|
|
|
class ValidationError(Exception): |
|
"""Custom exception for validation errors""" |
|
pass |
|
|
|
|
|
class ProcessingError(Exception): |
|
"""Custom exception for processing errors""" |
|
pass |
|
|
|
|
|
def handle_data_exceptions(func: Callable) -> Callable: |
|
""" |
|
Decorator to handle data-related exceptions |
|
""" |
|
@wraps(func) |
|
def wrapper(*args, **kwargs): |
|
try: |
|
return func(*args, **kwargs) |
|
except (DataError, ValidationError, ProcessingError) as e: |
|
logging.error(f"Data error in {func.__name__}: {str(e)}") |
|
st.error(f"Data error: {str(e)}") |
|
return None |
|
except FileNotFoundError as e: |
|
logging.error(f"File not found in {func.__name__}: {str(e)}") |
|
st.error(f"File not found: {str(e)}") |
|
return None |
|
except PermissionError as e: |
|
logging.error(f"Permission error in {func.__name__}: {str(e)}") |
|
st.error(f"Permission error: {str(e)}") |
|
return None |
|
except Exception as e: |
|
logging.error(f"Unexpected error in {func.__name__}: {str(e)}") |
|
logging.error(traceback.format_exc()) |
|
st.error(f"An unexpected error occurred: {str(e)}") |
|
return None |
|
return wrapper |
|
|
|
|
|
def handle_api_exceptions(func: Callable) -> Callable: |
|
""" |
|
Decorator to handle API-related exceptions |
|
""" |
|
@wraps(func) |
|
def wrapper(*args, **kwargs): |
|
try: |
|
return func(*args, **kwargs) |
|
except ConnectionError as e: |
|
logging.error(f"Connection error in {func.__name__}: {str(e)}") |
|
st.error("Connection error. Please check your internet connection.") |
|
return None |
|
except TimeoutError as e: |
|
logging.error(f"Timeout error in {func.__name__}: {str(e)}") |
|
st.error("Request timed out. Please try again.") |
|
return None |
|
except Exception as e: |
|
logging.error(f"API error in {func.__name__}: {str(e)}") |
|
st.error(f"API error: {str(e)}") |
|
return None |
|
return wrapper |
|
|
|
|
|
def log_error(message: str, error: Optional[Exception] = None): |
|
""" |
|
Log an error message with optional exception details |
|
""" |
|
if error: |
|
logging.error(f"{message}: {str(error)}") |
|
logging.error(traceback.format_exc()) |
|
else: |
|
logging.error(message) |
|
|
|
|
|
def display_error(message: str, error_type: str = "error"): |
|
""" |
|
Display an error message in the Streamlit interface |
|
""" |
|
if error_type == "warning": |
|
st.warning(message) |
|
elif error_type == "info": |
|
st.info(message) |
|
else: |
|
st.error(message) |
|
|
|
|
|
def validate_input(value: Any, value_type: type, field_name: str) -> bool: |
|
""" |
|
Validate input value and type |
|
""" |
|
if value is None: |
|
raise ValidationError(f"{field_name} cannot be None") |
|
|
|
if not isinstance(value, value_type): |
|
raise ValidationError(f"{field_name} must be of type {value_type.__name__}") |
|
|
|
return True |
|
|
|
|
|
def safe_execute(func: Callable, *args, **kwargs) -> tuple[bool, Any]: |
|
""" |
|
Safely execute a function and return success status and result |
|
""" |
|
try: |
|
result = func(*args, **kwargs) |
|
return True, result |
|
except Exception as e: |
|
log_error(f"Error executing {func.__name__}", e) |
|
return False, str(e) |
|
|
|
|
|
|
|
logging.basicConfig( |
|
level=logging.INFO, |
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', |
|
handlers=[ |
|
logging.FileHandler('app.log'), |
|
logging.StreamHandler() |
|
] |
|
) |