1"""
2Error Handling Strategies
3
4Demonstrates various error handling approaches:
51. Exception handling (try/except/finally)
62. Custom exceptions
73. Result pattern (type-safe error handling)
84. Context managers (resource management)
95. Retry logic with exponential backoff
106. Error recovery strategies
11"""
12
13from typing import Optional, TypeVar, Generic, Callable, Any
14from dataclasses import dataclass
15from enum import Enum
16import time
17import random
18from contextlib import contextmanager
19
20
21# =============================================================================
22# 1. EXCEPTION HANDLING BASICS
23# =============================================================================
24
25print("=" * 70)
26print("1. EXCEPTION HANDLING BASICS")
27print("=" * 70)
28
29
30def divide_with_error_handling(a: float, b: float) -> Optional[float]:
31 """Division with proper error handling"""
32 try:
33 result = a / b
34 return result
35 except ZeroDivisionError:
36 print(f"Error: Cannot divide {a} by zero")
37 return None
38 except TypeError as e:
39 print(f"Error: Invalid types - {e}")
40 return None
41 except Exception as e:
42 print(f"Unexpected error: {e}")
43 return None
44
45
46def read_file_with_cleanup(filename: str) -> Optional[str]:
47 """File reading with guaranteed cleanup using finally"""
48 file = None
49 try:
50 file = open(filename, 'r')
51 content = file.read()
52 return content
53 except FileNotFoundError:
54 print(f"Error: File '{filename}' not found")
55 return None
56 except PermissionError:
57 print(f"Error: Permission denied for '{filename}'")
58 return None
59 finally:
60 # Always executes, even if exception occurs
61 if file:
62 file.close()
63 print("File closed in finally block")
64
65
66# =============================================================================
67# 2. CUSTOM EXCEPTIONS
68# =============================================================================
69
70print("\n" + "=" * 70)
71print("2. CUSTOM EXCEPTIONS")
72print("=" * 70)
73
74
75class ApplicationError(Exception):
76 """Base exception for application errors"""
77 pass
78
79
80class ValidationError(ApplicationError):
81 """Raised when validation fails"""
82
83 def __init__(self, field: str, message: str):
84 self.field = field
85 self.message = message
86 super().__init__(f"Validation error in '{field}': {message}")
87
88
89class AuthenticationError(ApplicationError):
90 """Raised when authentication fails"""
91
92 def __init__(self, username: str):
93 self.username = username
94 super().__init__(f"Authentication failed for user '{username}'")
95
96
97class ResourceNotFoundError(ApplicationError):
98 """Raised when resource doesn't exist"""
99
100 def __init__(self, resource_type: str, resource_id: str):
101 self.resource_type = resource_type
102 self.resource_id = resource_id
103 super().__init__(
104 f"{resource_type} with id '{resource_id}' not found"
105 )
106
107
108class RateLimitError(ApplicationError):
109 """Raised when rate limit is exceeded"""
110
111 def __init__(self, limit: int, window: int):
112 self.limit = limit
113 self.window = window
114 super().__init__(
115 f"Rate limit exceeded: {limit} requests per {window} seconds"
116 )
117
118
119# Using custom exceptions
120def validate_user_age(age: int) -> None:
121 """Validate age with custom exception"""
122 if age < 0:
123 raise ValidationError("age", "Age cannot be negative")
124 if age < 18:
125 raise ValidationError("age", "Must be at least 18 years old")
126 if age > 150:
127 raise ValidationError("age", "Age seems unrealistic")
128
129
130# =============================================================================
131# 3. RESULT PATTERN (Type-safe error handling)
132# =============================================================================
133
134print("\n" + "=" * 70)
135print("3. RESULT PATTERN")
136print("=" * 70)
137
138T = TypeVar('T')
139E = TypeVar('E')
140
141
142@dataclass
143class Ok(Generic[T]):
144 """Represents successful result"""
145 value: T
146
147 def is_ok(self) -> bool:
148 return True
149
150 def is_err(self) -> bool:
151 return False
152
153 def unwrap(self) -> T:
154 return self.value
155
156 def unwrap_or(self, default: T) -> T:
157 return self.value
158
159 def map(self, func: Callable[[T], Any]) -> 'Result':
160 """Transform the value if Ok"""
161 try:
162 return Ok(func(self.value))
163 except Exception as e:
164 return Err(str(e))
165
166
167@dataclass
168class Err(Generic[E]):
169 """Represents error result"""
170 error: E
171
172 def is_ok(self) -> bool:
173 return False
174
175 def is_err(self) -> bool:
176 return True
177
178 def unwrap(self):
179 raise RuntimeError(f"Called unwrap on Err: {self.error}")
180
181 def unwrap_or(self, default):
182 return default
183
184 def map(self, func: Callable) -> 'Result':
185 """Do nothing if Err"""
186 return self
187
188
189# Type alias for Result
190Result = Ok[T] | Err[E]
191
192
193def divide_result(a: float, b: float) -> Result[float, str]:
194 """Division using Result pattern"""
195 if b == 0:
196 return Err("Division by zero")
197 return Ok(a / b)
198
199
200def parse_int_result(value: str) -> Result[int, str]:
201 """Parse integer using Result pattern"""
202 try:
203 return Ok(int(value))
204 except ValueError:
205 return Err(f"Cannot parse '{value}' as integer")
206
207
208def calculate_with_result(x: str, y: str) -> Result[float, str]:
209 """
210 Chain operations using Result pattern.
211 No exceptions thrown - all errors are values!
212 """
213 # Parse first number
214 x_result = parse_int_result(x)
215 if x_result.is_err():
216 return x_result
217
218 # Parse second number
219 y_result = parse_int_result(y)
220 if y_result.is_err():
221 return y_result
222
223 # Divide
224 return divide_result(float(x_result.unwrap()), float(y_result.unwrap()))
225
226
227# =============================================================================
228# 4. CONTEXT MANAGERS (Resource management)
229# =============================================================================
230
231print("\n" + "=" * 70)
232print("4. CONTEXT MANAGERS")
233print("=" * 70)
234
235
236class DatabaseConnection:
237 """Simulated database connection with context manager"""
238
239 def __init__(self, connection_string: str):
240 self.connection_string = connection_string
241 self.connected = False
242
243 def __enter__(self):
244 """Called when entering 'with' block"""
245 print(f"Opening connection to {self.connection_string}")
246 self.connected = True
247 return self
248
249 def __exit__(self, exc_type, exc_val, exc_tb):
250 """Called when exiting 'with' block (even if exception)"""
251 print(f"Closing connection to {self.connection_string}")
252 self.connected = False
253
254 # Return False to propagate exception, True to suppress
255 if exc_type is not None:
256 print(f"Exception occurred: {exc_type.__name__}: {exc_val}")
257 return False # Don't suppress exceptions
258
259 def query(self, sql: str) -> str:
260 if not self.connected:
261 raise RuntimeError("Not connected to database")
262 return f"Executing: {sql}"
263
264
265@contextmanager
266def timer(operation_name: str):
267 """Context manager for timing operations"""
268 start = time.time()
269 print(f"Starting {operation_name}...")
270
271 try:
272 yield # Code in 'with' block executes here
273 finally:
274 # Always executes, even if exception
275 elapsed = time.time() - start
276 print(f"{operation_name} took {elapsed:.4f} seconds")
277
278
279@contextmanager
280def error_handler(operation_name: str):
281 """Context manager for error handling"""
282 try:
283 yield
284 except Exception as e:
285 print(f"Error in {operation_name}: {e}")
286 # Could log, send alert, etc.
287 raise # Re-raise after handling
288
289
290# =============================================================================
291# 5. RETRY LOGIC
292# =============================================================================
293
294print("\n" + "=" * 70)
295print("5. RETRY LOGIC")
296print("=" * 70)
297
298
299def retry(
300 max_attempts: int = 3,
301 delay: float = 1.0,
302 backoff: float = 2.0,
303 exceptions: tuple = (Exception,)
304):
305 """
306 Decorator for retry logic with exponential backoff.
307
308 Args:
309 max_attempts: Maximum number of attempts
310 delay: Initial delay between retries (seconds)
311 backoff: Multiplier for delay after each retry
312 exceptions: Tuple of exceptions to catch
313 """
314
315 def decorator(func: Callable) -> Callable:
316 def wrapper(*args, **kwargs):
317 current_delay = delay
318 last_exception = None
319
320 for attempt in range(1, max_attempts + 1):
321 try:
322 return func(*args, **kwargs)
323 except exceptions as e:
324 last_exception = e
325 if attempt < max_attempts:
326 print(
327 f"Attempt {attempt} failed: {e}. "
328 f"Retrying in {current_delay:.2f}s..."
329 )
330 time.sleep(current_delay)
331 current_delay *= backoff
332 else:
333 print(f"All {max_attempts} attempts failed")
334
335 # All retries exhausted
336 raise last_exception
337
338 return wrapper
339
340 return decorator
341
342
343# Simulated unreliable function
344@retry(max_attempts=3, delay=0.1, backoff=2.0)
345def unreliable_network_call(success_rate: float = 0.3) -> str:
346 """Simulates unreliable network call"""
347 if random.random() > success_rate:
348 raise ConnectionError("Network timeout")
349 return "Success!"
350
351
352# =============================================================================
353# 6. ERROR RECOVERY STRATEGIES
354# =============================================================================
355
356print("\n" + "=" * 70)
357print("6. ERROR RECOVERY STRATEGIES")
358print("=" * 70)
359
360
361class ErrorRecoveryStrategy(Enum):
362 """Different error recovery strategies"""
363 FAIL_FAST = "fail_fast" # Fail immediately
364 RETURN_DEFAULT = "return_default" # Return default value
365 RETRY = "retry" # Retry operation
366 LOG_AND_CONTINUE = "log_and_continue" # Log error but continue
367
368
369def process_items(
370 items: list,
371 processor: Callable,
372 strategy: ErrorRecoveryStrategy = ErrorRecoveryStrategy.FAIL_FAST,
373 default_value: Any = None
374) -> list:
375 """
376 Process items with different error recovery strategies.
377 """
378 results = []
379
380 for i, item in enumerate(items):
381 try:
382 result = processor(item)
383 results.append(result)
384
385 except Exception as e:
386 if strategy == ErrorRecoveryStrategy.FAIL_FAST:
387 # Stop processing immediately
388 raise
389
390 elif strategy == ErrorRecoveryStrategy.RETURN_DEFAULT:
391 # Use default value for failed items
392 print(f"Item {i} failed: {e}, using default value")
393 results.append(default_value)
394
395 elif strategy == ErrorRecoveryStrategy.LOG_AND_CONTINUE:
396 # Log error and skip item
397 print(f"Item {i} failed: {e}, skipping")
398 continue
399
400 return results
401
402
403# =============================================================================
404# DEMONSTRATIONS
405# =============================================================================
406
407def demonstrate_basic_exceptions():
408 print("\n[BASIC EXCEPTION HANDLING]")
409 print("-" * 50)
410
411 # Normal division
412 result = divide_with_error_handling(10, 2)
413 print(f"10 / 2 = {result}")
414
415 # Division by zero
416 result = divide_with_error_handling(10, 0)
417 print(f"Result: {result}")
418
419
420def demonstrate_custom_exceptions():
421 print("\n[CUSTOM EXCEPTIONS]")
422 print("-" * 50)
423
424 try:
425 validate_user_age(25)
426 print("✓ Age 25 is valid")
427 except ValidationError as e:
428 print(f"✗ {e}")
429
430 try:
431 validate_user_age(15)
432 except ValidationError as e:
433 print(f"✗ {e} (field: {e.field})")
434
435 try:
436 raise ResourceNotFoundError("User", "12345")
437 except ResourceNotFoundError as e:
438 print(f"✗ {e}")
439
440
441def demonstrate_result_pattern():
442 print("\n[RESULT PATTERN]")
443 print("-" * 50)
444
445 # Successful division
446 result = divide_result(10, 2)
447 if result.is_ok():
448 print(f"Success: 10 / 2 = {result.unwrap()}")
449
450 # Division by zero (no exception!)
451 result = divide_result(10, 0)
452 if result.is_err():
453 print(f"Error: {result.error}")
454
455 # Use unwrap_or for default value
456 value = result.unwrap_or(0)
457 print(f"Value or default: {value}")
458
459 # Chaining operations
460 result = calculate_with_result("10", "2")
461 print(f"Calculate '10' / '2': {result.unwrap() if result.is_ok() else result.error}")
462
463 result = calculate_with_result("10", "abc")
464 print(f"Calculate '10' / 'abc': {result.error if result.is_err() else result.unwrap()}")
465
466
467def demonstrate_context_managers():
468 print("\n[CONTEXT MANAGERS]")
469 print("-" * 50)
470
471 # Database connection (resource cleanup guaranteed)
472 print("Normal usage:")
473 with DatabaseConnection("postgresql://localhost/mydb") as db:
474 print(db.query("SELECT * FROM users"))
475 print()
476
477 # Even with exception, connection is closed
478 print("With exception:")
479 try:
480 with DatabaseConnection("postgresql://localhost/mydb") as db:
481 print(db.query("SELECT * FROM users"))
482 raise ValueError("Something went wrong!")
483 except ValueError:
484 print("Exception caught, but connection was still closed\n")
485
486 # Timer context manager
487 with timer("Sleep operation"):
488 time.sleep(0.1)
489
490
491def demonstrate_retry():
492 print("\n[RETRY LOGIC]")
493 print("-" * 50)
494
495 # Will retry on failure
496 try:
497 result = unreliable_network_call(success_rate=0.5)
498 print(f"Result: {result}")
499 except ConnectionError:
500 print("Failed after all retries")
501
502
503def demonstrate_recovery_strategies():
504 print("\n[ERROR RECOVERY STRATEGIES]")
505 print("-" * 50)
506
507 items = [1, 2, "invalid", 4, 5]
508
509 def process(x):
510 return int(x) * 2
511
512 # Fail fast
513 print("Strategy: FAIL_FAST")
514 try:
515 results = process_items(items, process, ErrorRecoveryStrategy.FAIL_FAST)
516 except ValueError as e:
517 print(f"Stopped at error: {e}\n")
518
519 # Return default
520 print("Strategy: RETURN_DEFAULT")
521 results = process_items(
522 items, process,
523 ErrorRecoveryStrategy.RETURN_DEFAULT,
524 default_value=0
525 )
526 print(f"Results: {results}\n")
527
528 # Log and continue
529 print("Strategy: LOG_AND_CONTINUE")
530 results = process_items(items, process, ErrorRecoveryStrategy.LOG_AND_CONTINUE)
531 print(f"Results: {results}")
532
533
534def print_summary():
535 print("\n" + "=" * 70)
536 print("ERROR HANDLING STRATEGIES SUMMARY")
537 print("=" * 70)
538
539 print("""
5401. EXCEPTION HANDLING
541 ✓ Use try/except for expected errors
542 ✓ Use finally for cleanup
543 ✓ Catch specific exceptions
544 ✓ Don't catch Exception unless necessary
545
5462. CUSTOM EXCEPTIONS
547 ✓ Create hierarchy of exceptions
548 ✓ Add context (fields, messages)
549 ✓ Makes error handling more specific
550 ✓ Better than generic exceptions
551
5523. RESULT PATTERN
553 ✓ Errors as values (no exceptions)
554 ✓ Type-safe error handling
555 ✓ Explicit error handling required
556 ✓ Good for functional style
557
5584. CONTEXT MANAGERS
559 ✓ Guarantee resource cleanup
560 ✓ Use 'with' statement
561 ✓ Cleanup even with exceptions
562 ✓ Makes resource management safe
563
5645. RETRY LOGIC
565 ✓ Handle transient failures
566 ✓ Exponential backoff
567 ✓ Configurable attempts
568 ✓ Don't retry non-idempotent operations!
569
5706. RECOVERY STRATEGIES
571 ✓ Fail fast - stop immediately
572 ✓ Return default - continue with fallback
573 ✓ Log and continue - skip failed items
574 ✓ Choose based on requirements
575
576BEST PRACTICES:
577 • Be specific with exceptions
578 • Use context managers for resources
579 • Don't silently ignore errors
580 • Log errors appropriately
581 • Fail fast when appropriate
582 • Use Result pattern for predictable errors
583 • Always clean up resources
584 • Consider retry for transient failures
585""")
586
587
588if __name__ == "__main__":
589 demonstrate_basic_exceptions()
590 demonstrate_custom_exceptions()
591 demonstrate_result_pattern()
592 demonstrate_context_managers()
593 demonstrate_retry()
594 demonstrate_recovery_strategies()
595 print_summary()