Implement robust error handling and comprehensive logging for maintainable, debuggable applications.
Use specific exception handling with proper logging and meaningful error messages.
import logging
logger = logging.getLogger(__name__)
# ✅ Good exception handling
def fetch_user_profile(user_id: int) -> dict | None:
"""Fetch user profile with proper error handling."""
try:
# Validate input
if user_id <= 0:
raise ValueError(f"Invalid user ID: {user_id}")
# Database operation
profile = database.get_user_profile(user_id)
if profile is None:
logger.warning(f"User profile not found for ID: {user_id}")
return None
logger.info(f"Successfully fetched profile for user {user_id}")
return profile
except ValueError as e:
logger.error(f"Validation error in fetch_user_profile: {e}")
raise # Re-raise for caller to handle
except DatabaseConnectionError as e:
logger.error(f"Database connection failed: {e}")
raise ServiceUnavailableError("Unable to fetch user profile") from e
except DatabaseTimeoutError as e:
logger.error(f"Database timeout for user {user_id}: {e}")
raise ServiceUnavailableError("Request timed out") from e
except Exception as e:
logger.critical(f"Unexpected error in fetch_user_profile: {e}", exc_info=True)
raise InternalServerError("An unexpected error occurred") from e
# ✅ Custom exceptions with context
class UserServiceError(Exception):
"""Base exception for user service errors."""
pass
class UserNotFoundError(UserServiceError):
"""Raised when user cannot be found."""
def __init__(self, user_id: int):
self.user_id = user_id
super().__init__(f"User not found: {user_id}")
class DuplicateEmailError(UserServiceError):
"""Raised when email already exists."""
def __init__(self, email: str):
self.email = email
super().__init__(f"Email already exists: {email}")# ✅ Well-designed exception hierarchy
class APIError(Exception):
"""Base class for API-related errors."""
def __init__(self, message: str, error_code: int | None = None):
self.message = message
self.error_code = error_code
super().__init__(message)
class ValidationError(APIError):
"""Raised for input validation errors."""
def __init__(self, field: str, value: Any, message: str = None):
self.field = field
self.value = value
default_message = f"Invalid value '{value}' for field '{field}'"
super().__init__(message or default_message, error_code=400)
class AuthenticationError(APIError):
"""Raised for authentication failures."""
def __init__(self, message: str = "Authentication failed"):
super().__init__(message, error_code=401)
class AuthorizationError(APIError):
"""Raised for authorization failures."""
def __init__(self, resource: str, action: str):
message = f"Access denied: {action} on {resource}"
super().__init__(message, error_code=403)
self.resource = resource
self.action = action
class ResourceNotFoundError(APIError):
"""Raised when requested resource doesn't exist."""
def __init__(self, resource_type: str, identifier: str):
message = f"{resource_type} not found: {identifier}"
super().__init__(message, error_code=404)
self.resource_type = resource_type
self.identifier = identifier
# ✅ Using custom exceptions
def delete_user_account(user_id: int, requesting_user_id: int) -> bool:
"""Delete user account with proper authorization."""
try:
# Validate inputs
if user_id <= 0:
raise ValidationError("user_id", user_id, "User ID must be positive")
# Check if requesting user exists
requesting_user = get_user(requesting_user_id)
if not requesting_user:
raise AuthenticationError("Invalid requesting user")
# Check authorization
if not can_delete_user(requesting_user_id, user_id):
raise AuthorizationError("user_account", "delete")
# Check if target user exists
target_user = get_user(user_id)
if not target_user:
raise ResourceNotFoundError("User", str(user_id))
# Perform deletion
return database.delete_user(user_id)
except (ValidationError, AuthenticationError, AuthorizationError, ResourceNotFoundError):
# Re-raise API errors as-is
raise
except DatabaseError as e:
logger.error(f"Database error during user deletion: {e}")
raise APIError("Unable to delete user account", error_code=500) from eConfigure comprehensive logging with appropriate levels and structured formatting.
import logging
import logging.config
from datetime import datetime
# ✅ Comprehensive logging configuration
LOGGING_CONFIG = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'detailed': {
'format': '%(asctime)s [%(levelname)s] %(name)s:%(lineno)d - %(message)s',
'datefmt': '%Y-%m-%d %H:%M:%S'
},
'json': {
'format': '{"timestamp": "%(asctime)s", "level": "%(levelname)s", "logger": "%(name)s", "line": %(lineno)d, "message": "%(message)s"}',
'datefmt': '%Y-%m-%dT%H:%M:%S'
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'level': 'INFO',
'formatter': 'detailed',
'stream': 'ext://sys.stdout'
},
'file': {
'class': 'logging.handlers.RotatingFileHandler',
'level': 'DEBUG',
'formatter': 'detailed',
'filename': 'app.log',
'maxBytes': 10485760, # 10MB
'backupCount': 5
},
'error_file': {
'class': 'logging.handlers.RotatingFileHandler',
'level': 'ERROR',
'formatter': 'json',
'filename': 'errors.log',
'maxBytes': 10485760,
'backupCount': 10
}
},
'loggers': {
'': { # Root logger
'level': 'DEBUG',
'handlers': ['console', 'file', 'error_file']
},
'urllib3': {
'level': 'WARNING' # Reduce noise from requests
},
'boto3': {
'level': 'WARNING' # Reduce AWS SDK noise
}
}
}
# Initialize logging
logging.config.dictConfig(LOGGING_CONFIG)
logger = logging.getLogger(__name__)import logging
from functools import wraps
from time import time
logger = logging.getLogger(__name__)
# ✅ Structured logging with context
def process_payment(payment_id: str, amount: float, currency: str) -> dict:
"""Process payment with comprehensive logging."""
# Create correlation ID for tracking
correlation_id = generate_correlation_id()
# Log start with context
logger.info(
"Starting payment processing",
extra={
'correlation_id': correlation_id,
'payment_id': payment_id,
'amount': amount,
'currency': currency,
'action': 'payment_start'
}
)
start_time = time()
try:
# Validate payment
validate_payment(payment_id, amount, currency)
logger.debug(f"Payment validation successful: {payment_id}")
# Process with payment provider
result = payment_provider.charge(amount, currency)
# Log success
duration = time() - start_time
logger.info(
"Payment processed successfully",
extra={
'correlation_id': correlation_id,
'payment_id': payment_id,
'transaction_id': result.transaction_id,
'duration_seconds': duration,
'action': 'payment_success'
}
)
return {
'success': True,
'transaction_id': result.transaction_id,
'correlation_id': correlation_id
}
except ValidationError as e:
logger.warning(
f"Payment validation failed: {e}",
extra={
'correlation_id': correlation_id,
'payment_id': payment_id,
'error_type': 'validation',
'action': 'payment_failed'
}
)
raise
except PaymentProviderError as e:
logger.error(
f"Payment provider error: {e}",
extra={
'correlation_id': correlation_id,
'payment_id': payment_id,
'error_type': 'provider',
'error_code': e.error_code,
'action': 'payment_failed'
},
exc_info=True
)
raise
except Exception as e:
duration = time() - start_time
logger.critical(
f"Unexpected error in payment processing: {e}",
extra={
'correlation_id': correlation_id,
'payment_id': payment_id,
'duration_seconds': duration,
'error_type': 'unexpected',
'action': 'payment_failed'
},
exc_info=True
)
raise
# ✅ Logging decorator for function tracking
def log_function_call(logger_instance=None):
"""Decorator to log function calls with timing."""
def decorator(func):
nonlocal logger_instance
if logger_instance is None:
logger_instance = logging.getLogger(func.__module__)
@wraps(func)
def wrapper(*args, **kwargs):
func_name = f"{func.__module__}.{func.__name__}"
start_time = time()
logger_instance.debug(
f"Calling {func_name}",
extra={
'function': func_name,
'args_count': len(args),
'kwargs_count': len(kwargs)
}
)
try:
result = func(*args, **kwargs)
duration = time() - start_time
logger_instance.debug(
f"Completed {func_name}",
extra={
'function': func_name,
'duration_seconds': duration,
'success': True
}
)
return result
except Exception as e:
duration = time() - start_time
logger_instance.error(
f"Error in {func_name}: {e}",
extra={
'function': func_name,
'duration_seconds': duration,
'success': False,
'error_type': type(e).__name__
},
exc_info=True
)
raise
return wrapper
return decorator
# Usage
@log_function_call()
def complex_calculation(data: list[dict]) -> dict:
"""Perform complex calculation with automatic logging."""
# Function implementation
return {"result": "calculated"}import time
import random
from functools import wraps
from typing import Callable, TypeVar, Any
T = TypeVar('T')
# ✅ Retry decorator with exponential backoff
def retry_with_backoff(
max_attempts: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
exceptions: tuple = (Exception,)
):
"""Retry function with exponential backoff."""
def decorator(func: Callable[..., T]) -> Callable[..., T]:
@wraps(func)
def wrapper(*args, **kwargs) -> T:
last_exception = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt == max_attempts - 1:
logger.error(
f"All retry attempts failed for {func.__name__}",
extra={
'function': func.__name__,
'attempts': max_attempts,
'final_error': str(e)
}
)
raise
# Calculate delay with jitter
delay = min(base_delay * (2 ** attempt), max_delay)
jitter = delay * 0.1 * random.random()
actual_delay = delay + jitter
logger.warning(
f"Attempt {attempt + 1} failed for {func.__name__}, retrying in {actual_delay:.2f}s",
extra={
'function': func.__name__,
'attempt': attempt + 1,
'max_attempts': max_attempts,
'delay_seconds': actual_delay,
'error': str(e)
}
)
time.sleep(actual_delay)
# This should never be reached, but just in case
raise last_exception
return wrapper
return decorator
# ✅ Circuit breaker pattern
class CircuitBreaker:
"""Circuit breaker for handling external service failures."""
def __init__(
self,
failure_threshold: int = 5,
timeout: float = 60.0,
recovery_timeout: float = 30.0
):
self.failure_threshold = failure_threshold
self.timeout = timeout
self.recovery_timeout = recovery_timeout
self.failure_count = 0
self.last_failure_time = None
self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
def __call__(self, func: Callable[..., T]) -> Callable[..., T]:
@wraps(func)
def wrapper(*args, **kwargs) -> T:
if self.state == "OPEN":
if time.time() - self.last_failure_time < self.recovery_timeout:
raise CircuitBreakerOpenError(
f"Circuit breaker is OPEN for {func.__name__}"
)
else:
self.state = "HALF_OPEN"
logger.info(f"Circuit breaker entering HALF_OPEN state for {func.__name__}")
try:
result = func(*args, **kwargs)
# Success resets failure count
if self.state == "HALF_OPEN":
self.state = "CLOSED"
logger.info(f"Circuit breaker reset to CLOSED for {func.__name__}")
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
logger.error(
f"Circuit breaker recorded failure for {func.__name__}",
extra={
'function': func.__name__,
'failure_count': self.failure_count,
'threshold': self.failure_threshold,
'error': str(e)
}
)
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
logger.warning(
f"Circuit breaker opened for {func.__name__}",
extra={
'function': func.__name__,
'failure_count': self.failure_count
}
)
raise
return wrapper
class CircuitBreakerOpenError(Exception):
"""Raised when circuit breaker is in OPEN state."""
pass
# Usage examples
@retry_with_backoff(max_attempts=3, exceptions=(requests.RequestException,))
@CircuitBreaker(failure_threshold=5, recovery_timeout=30.0)
def call_external_api(endpoint: str) -> dict:
"""Call external API with retry and circuit breaker protection."""
response = requests.get(endpoint, timeout=10)
response.raise_for_status()
return response.json()