Python - Custom Exceptions
• Custom exceptions create a semantic layer in your code that makes error handling explicit and maintainable, replacing generic exceptions with domain-specific error types that communicate intent
Key Insights
• Custom exceptions create a semantic layer in your code that makes error handling explicit and maintainable, replacing generic exceptions with domain-specific error types that communicate intent • Exception hierarchies enable granular error handling where you can catch broad categories of errors or specific cases, while maintaining consistent error attributes and behavior across your application • Proper exception design includes meaningful error messages, relevant context data, and appropriate inheritance structures that align with your application’s domain model
Why Custom Exceptions Matter
Python’s built-in exceptions cover general programming errors, but they don’t express your application’s business logic failures. When you raise ValueError for an invalid user ID, an out-of-range price, and a malformed email address, you lose semantic meaning and force callers to parse error messages to distinguish between cases.
Custom exceptions transform error handling from string parsing to type-based flow control:
# Poor approach - generic exceptions
def process_payment(amount, user_id):
if amount <= 0:
raise ValueError("Invalid amount")
if not user_exists(user_id):
raise ValueError("Invalid user")
# Processing logic
# Better approach - custom exceptions
def process_payment(amount, user_id):
if amount <= 0:
raise InvalidPaymentAmountError(amount)
if not user_exists(user_id):
raise UserNotFoundError(user_id)
# Processing logic
The second version lets callers handle specific failures differently without inspecting error messages.
Basic Custom Exception Pattern
Start with the simplest pattern: inherit from Exception or a more specific built-in exception class:
class ApplicationError(Exception):
"""Base exception for all application errors"""
pass
class ValidationError(ApplicationError):
"""Raised when data validation fails"""
pass
class ResourceNotFoundError(ApplicationError):
"""Raised when a requested resource doesn't exist"""
pass
# Usage
def get_user(user_id):
user = db.query(User).filter_by(id=user_id).first()
if not user:
raise ResourceNotFoundError(f"User {user_id} not found")
return user
try:
user = get_user(123)
except ResourceNotFoundError as e:
logger.warning(str(e))
return None
except ApplicationError as e:
logger.error(f"Application error: {e}")
raise
This establishes a hierarchy where catching ApplicationError catches all custom exceptions, while specific handlers catch individual cases.
Adding Context and Attributes
Exceptions should carry structured data, not just strings. Add attributes that callers can inspect programmatically:
class PaymentError(ApplicationError):
"""Base class for payment-related errors"""
def __init__(self, message, amount=None, transaction_id=None):
super().__init__(message)
self.amount = amount
self.transaction_id = transaction_id
self.timestamp = datetime.utcnow()
class InsufficientFundsError(PaymentError):
def __init__(self, required, available, account_id):
message = f"Insufficient funds: required {required}, available {available}"
super().__init__(message, amount=required)
self.required = required
self.available = available
self.account_id = account_id
self.shortfall = required - available
# Usage with structured error handling
try:
process_payment(amount=500, account_id="ACC123")
except InsufficientFundsError as e:
logger.warning(
"Payment failed",
extra={
"account_id": e.account_id,
"shortfall": e.shortfall,
"timestamp": e.timestamp
}
)
notify_user_insufficient_funds(e.account_id, e.shortfall)
except PaymentError as e:
handle_payment_error(e)
Exception Hierarchies for API Design
Design exception hierarchies that mirror your domain model. This enables flexible error handling at different abstraction levels:
class DatabaseError(ApplicationError):
"""Database operation failures"""
pass
class ConnectionError(DatabaseError):
"""Database connection failures"""
pass
class QueryError(DatabaseError):
"""Query execution failures"""
def __init__(self, query, params=None, original_error=None):
self.query = query
self.params = params
self.original_error = original_error
message = f"Query failed: {query[:100]}"
if original_error:
message += f" ({type(original_error).__name__}: {original_error})"
super().__init__(message)
class IntegrityError(QueryError):
"""Database constraint violations"""
pass
class DuplicateRecordError(IntegrityError):
"""Unique constraint violations"""
def __init__(self, table, field, value):
self.table = table
self.field = field
self.value = value
query = f"INSERT INTO {table}"
super().__init__(
query=query,
params={field: value}
)
# Granular handling
try:
create_user(email="existing@example.com")
except DuplicateRecordError as e:
return {"error": f"Email {e.value} already registered"}
except IntegrityError as e:
return {"error": "Data validation failed"}
except DatabaseError as e:
logger.error(f"Database error: {e}")
return {"error": "Service temporarily unavailable"}
Validation Exception Patterns
For data validation, create exceptions that aggregate multiple validation failures:
class ValidationError(ApplicationError):
"""Single validation failure"""
def __init__(self, field, message, value=None):
self.field = field
self.value = value
super().__init__(f"{field}: {message}")
class ValidationErrors(ApplicationError):
"""Multiple validation failures"""
def __init__(self, errors):
self.errors = errors # List of ValidationError instances
messages = [str(e) for e in errors]
super().__init__(f"Validation failed: {'; '.join(messages)}")
def to_dict(self):
"""Convert to API-friendly format"""
return {
error.field: str(error)
for error in self.errors
}
def validate_user_data(data):
errors = []
if not data.get('email') or '@' not in data['email']:
errors.append(ValidationError('email', 'Invalid email format', data.get('email')))
if not data.get('age') or data['age'] < 18:
errors.append(ValidationError('age', 'Must be 18 or older', data.get('age')))
if len(data.get('password', '')) < 8:
errors.append(ValidationError('password', 'Must be at least 8 characters'))
if errors:
raise ValidationErrors(errors)
return data
# API endpoint usage
@app.post('/users')
def create_user(data):
try:
validated_data = validate_user_data(data)
user = User.create(**validated_data)
return {"id": user.id}, 201
except ValidationErrors as e:
return {"errors": e.to_dict()}, 400
Retry and Recovery Patterns
Custom exceptions enable sophisticated retry logic by distinguishing transient from permanent failures:
class RetryableError(ApplicationError):
"""Errors that may succeed on retry"""
pass
class PermanentError(ApplicationError):
"""Errors that won't succeed on retry"""
pass
class RateLimitError(RetryableError):
def __init__(self, retry_after):
self.retry_after = retry_after
super().__init__(f"Rate limited. Retry after {retry_after} seconds")
class AuthenticationError(PermanentError):
"""Invalid credentials - retry won't help"""
pass
def retry_with_backoff(func, max_attempts=3):
for attempt in range(max_attempts):
try:
return func()
except RateLimitError as e:
if attempt == max_attempts - 1:
raise
time.sleep(e.retry_after)
except RetryableError as e:
if attempt == max_attempts - 1:
raise
time.sleep(2 ** attempt) # Exponential backoff
except PermanentError:
raise # Don't retry permanent failures
# Usage
try:
result = retry_with_backoff(lambda: api_call())
except PermanentError as e:
logger.error(f"Permanent failure: {e}")
raise
except RetryableError as e:
logger.error(f"Failed after retries: {e}")
raise
Context Managers and Exception Chaining
Use exception chaining to preserve the original error while adding context:
class ConfigurationError(ApplicationError):
"""Configuration loading or validation failures"""
pass
def load_config(path):
try:
with open(path) as f:
return json.load(f)
except FileNotFoundError as e:
raise ConfigurationError(
f"Configuration file not found: {path}"
) from e
except json.JSONDecodeError as e:
raise ConfigurationError(
f"Invalid JSON in configuration: {e.msg} at line {e.lineno}"
) from e
# The 'from e' preserves the original exception
try:
config = load_config('config.json')
except ConfigurationError as e:
logger.error(f"Config error: {e}")
logger.debug(f"Original error: {e.__cause__}")
raise
Testing Custom Exceptions
Test both that exceptions are raised correctly and that they carry the expected data:
import pytest
def test_insufficient_funds_error():
with pytest.raises(InsufficientFundsError) as exc_info:
raise InsufficientFundsError(
required=100,
available=50,
account_id="ACC123"
)
error = exc_info.value
assert error.required == 100
assert error.available == 50
assert error.shortfall == 50
assert "ACC123" in str(error)
def test_validation_errors_aggregation():
errors = [
ValidationError('email', 'Invalid format'),
ValidationError('age', 'Too young')
]
exc = ValidationErrors(errors)
error_dict = exc.to_dict()
assert 'email' in error_dict
assert 'age' in error_dict
assert len(error_dict) == 2
Custom exceptions transform error handling from an afterthought into a first-class design element. They make your code’s failure modes explicit, enable precise error recovery strategies, and provide structured data for logging and monitoring systems.