python
exercises
exercises.py🐍python
"""
08 - Error Handling: Exercises
Complete these exercises to master exception handling!
"""
from typing import Optional, TypeVar, Callable, Any
import time
import logging
T = TypeVar('T')
# =============================================================================
# EXERCISE 1: Safe Type Conversion
# =============================================================================
def safe_int(value: Any, default: int = 0) -> int:
"""
Safely convert a value to int, returning default on failure.
Args:
value: Value to convert
default: Value to return if conversion fails
Returns:
Converted int or default
"""
# YOUR CODE HERE
try:
return int(value)
except (ValueError, TypeError):
return default
def safe_float(value: Any, default: float = 0.0) -> float:
"""
Safely convert a value to float, returning default on failure.
Args:
value: Value to convert
default: Value to return if conversion fails
Returns:
Converted float or default
"""
# YOUR CODE HERE
try:
return float(value)
except (ValueError, TypeError):
return default
def safe_get(dictionary: dict, key: str, default: Any = None) -> Any:
"""
Safely get a value from a dictionary.
Args:
dictionary: Dictionary to get from
key: Key to look up
default: Value to return if key not found
Returns:
Value at key or default
"""
# YOUR CODE HERE
try:
return dictionary[key]
except (KeyError, TypeError):
return default
# =============================================================================
# EXERCISE 2: Input Validation
# =============================================================================
class ValidationError(Exception):
"""Custom exception for validation errors."""
pass
def validate_email(email: str) -> str:
"""
Validate an email address (basic check).
Requirements:
- Must be a string
- Must contain exactly one @
- Must have at least one character before @
- Must have at least one . after @
Args:
email: Email address to validate
Returns:
Validated email (stripped and lowercase)
Raises:
ValidationError: If email is invalid
"""
# YOUR CODE HERE
if not isinstance(email, str):
raise ValidationError("Email must be a string")
email = email.strip().lower()
if email.count('@') != 1:
raise ValidationError("Email must contain exactly one @")
local, domain = email.split('@')
if len(local) == 0:
raise ValidationError("Email must have characters before @")
if '.' not in domain:
raise ValidationError("Email domain must contain a dot")
return email
def validate_password(password: str, min_length: int = 8) -> str:
"""
Validate a password.
Requirements:
- At least min_length characters
- At least one uppercase letter
- At least one lowercase letter
- At least one digit
Args:
password: Password to validate
min_length: Minimum required length
Returns:
The validated password
Raises:
ValidationError: If password is invalid
"""
# YOUR CODE HERE
if not isinstance(password, str):
raise ValidationError("Password must be a string")
if len(password) < min_length:
raise ValidationError(f"Password must be at least {min_length} characters")
if not any(c.isupper() for c in password):
raise ValidationError("Password must contain an uppercase letter")
if not any(c.islower() for c in password):
raise ValidationError("Password must contain a lowercase letter")
if not any(c.isdigit() for c in password):
raise ValidationError("Password must contain a digit")
return password
# =============================================================================
# EXERCISE 3: Custom Exception Hierarchy
# =============================================================================
class AppError(Exception):
"""Base exception for application errors."""
def __init__(self, message: str, code: str = "UNKNOWN"):
self.message = message
self.code = code
super().__init__(message)
class NetworkError(AppError):
"""Exception for network-related errors."""
def __init__(self, message: str, url: Optional[str] = None):
super().__init__(message, "NETWORK_ERROR")
self.url = url
class AuthenticationError(AppError):
"""Exception for authentication failures."""
def __init__(self, message: str, username: Optional[str] = None):
super().__init__(message, "AUTH_ERROR")
self.username = username
class DataError(AppError):
"""Exception for data-related errors."""
def __init__(self, message: str, field: Optional[str] = None):
super().__init__(message, "DATA_ERROR")
self.field = field
def handle_api_error(error: AppError) -> dict:
"""
Convert an AppError to a standard API response format.
Args:
error: The application error
Returns:
Dict with 'success', 'error_code', and 'message' keys
"""
# YOUR CODE HERE
return {
'success': False,
'error_code': error.code,
'message': error.message
}
# =============================================================================
# EXERCISE 4: Retry Decorator
# =============================================================================
def retry(max_attempts: int = 3, delay: float = 1.0, exceptions: tuple = (Exception,)):
"""
Decorator that retries a function on failure.
Args:
max_attempts: Maximum number of attempts
delay: Seconds to wait between attempts
exceptions: Tuple of exception types to catch
Returns:
Decorator function
"""
def decorator(func: Callable[..., T]) -> Callable[..., T]:
def wrapper(*args: Any, **kwargs: Any) -> T:
# YOUR CODE HERE
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(delay)
raise last_exception # type: ignore
return wrapper
return decorator
# =============================================================================
# EXERCISE 5: Error Collector
# =============================================================================
class ErrorCollector:
"""
Collect errors during batch processing without stopping.
Usage:
collector = ErrorCollector()
for item in items:
with collector.catch():
process(item)
if collector.has_errors:
handle_errors(collector.errors)
"""
def __init__(self):
self.errors: list[tuple[Exception, str]] = []
@property
def has_errors(self) -> bool:
"""Return True if any errors were collected."""
return len(self.errors) > 0
def add_error(self, error: Exception, context: str = "") -> None:
"""Add an error to the collection."""
self.errors.append((error, context))
class _CatchContext:
"""Context manager for catching errors."""
def __init__(self, collector: 'ErrorCollector', context: str):
self.collector = collector
self.context = context
def __enter__(self) -> None:
pass
def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
if exc_val is not None:
self.collector.add_error(exc_val, self.context)
return True # Suppress exception
return False
def catch(self, context: str = "") -> '_CatchContext':
"""
Return a context manager that catches and collects errors.
Args:
context: Description of what was being attempted
Returns:
Context manager
"""
# YOUR CODE HERE
return self._CatchContext(self, context)
def summary(self) -> str:
"""
Return a summary of all collected errors.
Returns:
Multi-line string describing all errors
"""
# YOUR CODE HERE
if not self.has_errors:
return "No errors collected"
lines = [f"Collected {len(self.errors)} error(s):"]
for i, (error, context) in enumerate(self.errors, 1):
ctx = f" ({context})" if context else ""
lines.append(f" {i}. {type(error).__name__}: {error}{ctx}")
return "\n".join(lines)
# =============================================================================
# EXERCISE 6: Resource Manager
# =============================================================================
class DatabaseConnection:
"""Simulated database connection."""
def __init__(self, connection_string: str):
self.connection_string = connection_string
self.connected = False
self.transaction_active = False
def connect(self) -> None:
"""Establish connection."""
if "invalid" in self.connection_string:
raise ConnectionError("Invalid connection string")
self.connected = True
def disconnect(self) -> None:
"""Close connection."""
if self.transaction_active:
self.rollback()
self.connected = False
def begin_transaction(self) -> None:
"""Start a transaction."""
if not self.connected:
raise RuntimeError("Not connected")
self.transaction_active = True
def commit(self) -> None:
"""Commit the transaction."""
if not self.transaction_active:
raise RuntimeError("No active transaction")
self.transaction_active = False
def rollback(self) -> None:
"""Rollback the transaction."""
self.transaction_active = False
def execute(self, query: str) -> str:
"""Execute a query."""
if not self.connected:
raise RuntimeError("Not connected")
if "error" in query.lower():
raise RuntimeError(f"Query failed: {query}")
return f"Result for: {query}"
class TransactionManager:
"""
Context manager for database transactions.
Automatically commits on success, rolls back on failure.
"""
def __init__(self, connection: DatabaseConnection):
self.connection = connection
def __enter__(self) -> DatabaseConnection:
"""Start transaction and return connection."""
# YOUR CODE HERE
self.connection.begin_transaction()
return self.connection
def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
"""Commit or rollback based on success."""
# YOUR CODE HERE
if exc_type is not None:
self.connection.rollback()
return False # Propagate exception
else:
self.connection.commit()
return False
# =============================================================================
# EXERCISE 7: Error Wrapper
# =============================================================================
def wrap_errors(wrapper_exception: type, *exception_types: type):
"""
Decorator that wraps specified exceptions in a wrapper exception.
Args:
wrapper_exception: Exception type to wrap errors in
exception_types: Exception types to catch and wrap
Returns:
Decorator function
"""
def decorator(func: Callable[..., T]) -> Callable[..., T]:
def wrapper(*args: Any, **kwargs: Any) -> T:
# YOUR CODE HERE
try:
return func(*args, **kwargs)
except exception_types as e:
raise wrapper_exception(str(e)) from e
return wrapper
return decorator
# =============================================================================
# EXERCISE 8: Graceful Degradation
# =============================================================================
def with_fallback(primary_func: Callable[..., T],
fallback_func: Callable[..., T]) -> Callable[..., T]:
"""
Create a function that falls back to another on failure.
Args:
primary_func: Main function to try
fallback_func: Backup function if primary fails
Returns:
A function that tries primary, then fallback
"""
# YOUR CODE HERE
def wrapper(*args: Any, **kwargs: Any) -> T:
try:
return primary_func(*args, **kwargs)
except Exception:
return fallback_func(*args, **kwargs)
return wrapper
# =============================================================================
# EXERCISE 9: Error Rate Limiter
# =============================================================================
class CircuitBreaker:
"""
Circuit breaker pattern - stops calling a function after too many failures.
States:
- CLOSED: Normal operation, calls pass through
- OPEN: Too many failures, calls fail immediately
- HALF_OPEN: Testing if service recovered
"""
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
def __init__(self, failure_threshold: int = 5, reset_timeout: float = 30.0):
self.failure_threshold = failure_threshold
self.reset_timeout = reset_timeout
self.failures = 0
self.state = self.CLOSED
self.last_failure_time: Optional[float] = None
def _check_state(self) -> None:
"""Update state based on timeout."""
if self.state == self.OPEN and self.last_failure_time:
if time.time() - self.last_failure_time > self.reset_timeout:
self.state = self.HALF_OPEN
def call(self, func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
"""
Call function through the circuit breaker.
Args:
func: Function to call
*args, **kwargs: Arguments to pass
Returns:
Result from function
Raises:
RuntimeError: If circuit is open
Exception: If function fails
"""
# YOUR CODE HERE
self._check_state()
if self.state == self.OPEN:
raise RuntimeError("Circuit breaker is open")
try:
result = func(*args, **kwargs)
# Success - reset failures
if self.state == self.HALF_OPEN:
self.state = self.CLOSED
self.failures = 0
return result
except Exception as e:
self.failures += 1
self.last_failure_time = time.time()
if self.failures >= self.failure_threshold:
self.state = self.OPEN
raise
def reset(self) -> None:
"""Manually reset the circuit breaker."""
self.failures = 0
self.state = self.CLOSED
self.last_failure_time = None
# =============================================================================
# EXERCISE 10: Comprehensive Error Handler
# =============================================================================
class Result:
"""
Result type that can be either success or failure.
Similar to Rust's Result or functional programming patterns.
"""
def __init__(self, value: Optional[T] = None, error: Optional[Exception] = None):
self._value = value
self._error = error
@classmethod
def ok(cls, value: T) -> 'Result':
"""Create a success result."""
return cls(value=value)
@classmethod
def err(cls, error: Exception) -> 'Result':
"""Create a failure result."""
return cls(error=error)
@property
def is_ok(self) -> bool:
"""Return True if result is success."""
return self._error is None
@property
def is_err(self) -> bool:
"""Return True if result is failure."""
return self._error is not None
def unwrap(self) -> T:
"""
Get the value, raising if error.
Returns:
The success value
Raises:
The stored exception if result is an error
"""
# YOUR CODE HERE
if self._error is not None:
raise self._error
return self._value # type: ignore
def unwrap_or(self, default: T) -> T:
"""
Get the value or return default if error.
Args:
default: Value to return if result is error
Returns:
The success value or default
"""
# YOUR CODE HERE
if self._error is not None:
return default
return self._value # type: ignore
def map(self, func: Callable[[T], Any]) -> 'Result':
"""
Apply a function to the value if success.
Args:
func: Function to apply
Returns:
New Result with transformed value or same error
"""
# YOUR CODE HERE
if self._error is not None:
return Result.err(self._error)
try:
return Result.ok(func(self._value))
except Exception as e:
return Result.err(e)
def try_operation(func: Callable[..., T]) -> Callable[..., Result]:
"""
Decorator that wraps function return in Result type.
Args:
func: Function to wrap
Returns:
Wrapped function that returns Result
"""
def wrapper(*args: Any, **kwargs: Any) -> Result:
try:
value = func(*args, **kwargs)
return Result.ok(value)
except Exception as e:
return Result.err(e)
return wrapper
# =============================================================================
# TEST YOUR SOLUTIONS
# =============================================================================
if __name__ == "__main__":
print("Testing error handling exercises...")
print("=" * 60)
# Test 1: Safe conversion
assert safe_int("42") == 42
assert safe_int("abc") == 0
assert safe_int("abc", -1) == -1
print("✓ safe_int")
assert safe_float("3.14") == 3.14
assert safe_float("abc") == 0.0
print("✓ safe_float")
assert safe_get({"a": 1}, "a") == 1
assert safe_get({"a": 1}, "b", "default") == "default"
print("✓ safe_get")
# Test 2: Validation
assert validate_email("TEST@example.COM") == "test@example.com"
try:
validate_email("invalid")
assert False, "Should raise"
except ValidationError:
pass
print("✓ validate_email")
assert validate_password("Secure123")
try:
validate_password("weak")
assert False, "Should raise"
except ValidationError:
pass
print("✓ validate_password")
# Test 3: Custom exceptions
error = NetworkError("Connection failed", url="http://example.com")
response = handle_api_error(error)
assert response['success'] is False
assert response['error_code'] == "NETWORK_ERROR"
print("✓ Custom exceptions")
# Test 4: Retry decorator
attempt_count = 0
@retry(max_attempts=3, delay=0.01)
def flaky():
global attempt_count
attempt_count += 1
if attempt_count < 3:
raise ValueError("Not yet")
return "success"
result = flaky()
assert result == "success"
assert attempt_count == 3
print("✓ retry decorator")
# Test 5: Error collector
collector = ErrorCollector()
items = [1, 2, "three", 4, "five"]
results = []
for item in items:
with collector.catch(f"Processing {item}"):
results.append(int(item))
assert results == [1, 2, 4]
assert collector.has_errors
assert len(collector.errors) == 2
print("✓ ErrorCollector")
# Test 6: Transaction manager
conn = DatabaseConnection("test://localhost")
conn.connect()
with TransactionManager(conn) as db:
db.execute("SELECT * FROM users")
assert not conn.transaction_active
print("✓ TransactionManager (success)")
try:
with TransactionManager(conn) as db:
db.execute("SELECT * FROM error")
except RuntimeError:
pass
assert not conn.transaction_active # Should have rolled back
print("✓ TransactionManager (rollback)")
conn.disconnect()
# Test 7: Error wrapper
@wrap_errors(AppError, ValueError, TypeError)
def fragile_func(x):
if x < 0:
raise ValueError("Negative!")
return x * 2
try:
fragile_func(-1)
except AppError as e:
assert "Negative" in str(e)
print("✓ wrap_errors")
# Test 8: Fallback
def primary(x):
raise RuntimeError("Primary failed")
def backup(x):
return x * 10
safe_func = with_fallback(primary, backup)
assert safe_func(5) == 50
print("✓ with_fallback")
# Test 9: Circuit breaker
breaker = CircuitBreaker(failure_threshold=2, reset_timeout=0.1)
fail_count = 0
def unstable():
global fail_count
fail_count += 1
if fail_count <= 3:
raise ValueError("Failure!")
return "OK"
# First two failures
for _ in range(2):
try:
breaker.call(unstable)
except ValueError:
pass
# Should be open now
try:
breaker.call(unstable)
assert False, "Should raise RuntimeError"
except RuntimeError as e:
assert "open" in str(e).lower()
print("✓ CircuitBreaker")
# Test 10: Result type
result = Result.ok(42)
assert result.is_ok
assert result.unwrap() == 42
print("✓ Result.ok")
result = Result.err(ValueError("oops"))
assert result.is_err
assert result.unwrap_or(0) == 0
print("✓ Result.err")
result = Result.ok(5).map(lambda x: x * 2)
assert result.unwrap() == 10
print("✓ Result.map")
@try_operation
def divide(a, b):
return a / b
result = divide(10, 2)
assert result.is_ok
assert result.unwrap() == 5
result = divide(10, 0)
assert result.is_err
print("✓ try_operation")
print("\n" + "=" * 60)
print("All tests passed! ✓")