Introduction
Robust error handling distinguishes production-ready code from prototypes. Applications must handle failures gracefully, provide meaningful feedback, and recover when possible. This guide covers error handling patterns for building resilient applications.
Exception Hierarchies
from abc import ABC
from typing import Optional
import logging
class AppError(Exception):
"""Base application error."""
def __init__(self, message: str, code: str = None):
super().__init__(message)
self.message = message
self.code = code or "INTERNAL_ERROR"
self.logger = logging.getLogger(__name__)
class ValidationError(AppError):
"""Invalid input data."""
def __init__(self, message: str, field: str = None):
super().__init__(message, "VALIDATION_ERROR")
self.field = field
class NotFoundError(AppError):
"""Resource not found."""
def __init__(self, resource: str, identifier: str):
super().__init__(f"{resource} not found: {identifier}", "NOT_FOUND")
self.resource = resource
self.identifier = identifier
class AuthenticationError(AppError):
"""Authentication failed."""
def __init__(self, message: str = "Authentication required"):
super().__init__(message, "AUTH_ERROR")
class AuthorizationError(AppError):
"""Permission denied."""
def __init__(self, message: str = "Permission denied"):
super().__init__(message, "FORBIDDEN")
class ExternalServiceError(AppError):
"""External service failure."""
def __init__(self, service: str, message: str):
super().__init__(message, "EXTERNAL_SERVICE_ERROR")
self.service = service
# Usage
def get_user(user_id: str) -> User:
user = user_repo.find_by_id(user_id)
if not user:
raise NotFoundError("User", user_id)
return user
Result Types
from dataclasses import dataclass
from typing import TypeVar, Generic, Optional
T = TypeVar('T')
@dataclass
class Result(Generic[T]):
"""Result type for operations that can fail."""
_value: Optional[T]
_error: Optional[Exception]
@classmethod
def success(cls, value: T) -> 'Result[T]':
return cls(_value=value, _error=None)
@classmethod
def failure(cls, error: Exception) -> 'Result[T]':
return cls(_value=None, _error=error)
@property
def is_success(self) -> bool:
return self._error is None
@property
def is_failure(self) -> bool:
return self._error is not None
def get_or_none(self) -> Optional[T]:
return self._value
def get_or_raise(self) -> T:
if self._error:
raise self._error
return self._value
def get_or_default(self, default: T) -> T:
return self._value if self._value is not None else default
def map(self, func) -> 'Result':
if self.is_success:
try:
return Result.success(func(self._value))
except Exception as e:
return Result.failure(e)
return self
# Usage
def create_user(email: str) -> Result[User]:
try:
if not validate_email(email):
return Result.failure(ValidationError("Invalid email"))
if user_repo.find_by_email(email):
return Result.failure(ValidationError("Email exists"))
user = User(email)
user_repo.save(user)
return Result.success(user)
except Exception as e:
return Result.failure(e)
result = create_user("[email protected]")
if result.is_success:
user = result.get_or_raise()
else:
log_error(result._error)
Retry Logic
import time
from functools import wraps
from typing import Callable, Type
def retry(
max_attempts: int = 3,
delay: float = 1.0,
backoff: float = 2.0,
exceptions: tuple = (Exception,)
):
def decorator(func: Callable):
@wraps(func)
def wrapper(*args, **kwargs):
current_delay = delay
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:
time.sleep(current_delay)
current_delay *= backoff
raise last_exception
return wrapper
return decorator
# Usage
@retry(max_attempts=3, delay=0.5, exceptions=(ExternalServiceError,))
def call_external_api(endpoint: str) -> dict:
response = requests.get(endpoint)
if response.status_code >= 500:
raise ExternalServiceError("api", f"Status {response.status_code}")
return response.json()
Circuit Breaker
from enum import Enum
from datetime import datetime, timedelta
import time
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class CircuitBreaker:
"""Circuit breaker for preventing cascade failures."""
def __init__(
self,
name: str,
failure_threshold: int = 5,
recovery_timeout: float = 60.0
):
self.name = name
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self._state = CircuitState.CLOSED
self._failure_count = 0
self._last_failure_time = None
self._success_count = 0
@property
def state(self) -> CircuitState:
if self._state == CircuitState.OPEN:
if time.time() - self._last_failure_time > self.recovery_timeout:
self._state = CircuitState.HALF_OPEN
self._success_count = 0
return self._state
def call(self, func, *args, **kwargs):
if self.state == CircuitState.OPEN:
raise CircuitOpenError(f"Circuit {self.name} is open")
try:
result = func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _on_success(self):
if self._state == CircuitState.HALF_OPEN:
self._success_count += 1
if self._success_count >= 3:
self._state = CircuitState.CLOSED
self._failure_count = 0
else:
self._failure_count = 0
def _on_failure(self):
self._failure_count += 1
self._last_failure_time = time.time()
if self._failure_count >= self.failure_threshold:
self._state = CircuitState.OPEN
# Usage
breaker = CircuitBreaker("payment-service", failure_threshold=3)
def process_payment(order_id: str) -> Payment:
return breaker.call(payment_gateway.charge, order_id)
Conclusion
Effective error handling requires strategy: define exception hierarchies, use result types for recoverable errors, implement retry with backoff, and use circuit breakers to prevent cascade failures. Always log errors with context and provide meaningful messages to users.
Resources
- “Release It!” by Michael T. Nygard
- Microsoft Error Handling Guidelines
Comments