Master Python error handling, debugging techniques, and building robust applications that gracefully handle unexpected situations and provide meaningful error information.
Exception handling and debugging are critical skills that separate amateur programmers from professionals. In production environments, applications must gracefully handle errors, provide meaningful feedback, and continue operating reliably even when unexpected situations occur.
Python provides a comprehensive exception handling system that allows you to anticipate, catch, and respond to errors appropriately. Combined with powerful debugging tools, these skills enable you to build robust applications that users can trust and maintain easily.
Companies like Google, Netflix, and Airbnb rely on sophisticated error handling and monitoring to ensure their applications remain stable under any conditions. In this lesson, you'll learn professional-grade techniques for handling errors and debugging Python applications effectively.
Python has a well-organized exception hierarchy that helps you understand and handle different types of errors systematically.
Common Built-in Exceptions:
1# ValueError - Invalid value for operation
2try:
3 number = int("not_a_number")
4except ValueError as e:
5 print(f"ValueError: {e}")
6
7# TypeError - Wrong type for operation
8try:
9 result = "hello" + 5
10except TypeError as e:
11 print(f"TypeError: {e}")
12
13# KeyError - Key not found in dictionary
14try:
15 data = {"name": "Alice"}
16 age = data["age"]
17except KeyError as e:
18 print(f"KeyError: Missing key {e}")
19
20# IndexError - Index out of range
21try:
22 items = [1, 2, 3]
23 item = items[10]
24except IndexError as e:
25 print(f"IndexError: {e}")
26
27# FileNotFoundError - File doesn't exist
28try:
29 with open("nonexistent.txt", "r") as file:
30 content = file.read()
31except FileNotFoundError as e:
32 print(f"FileNotFoundError: {e}")
33
34# AttributeError - Attribute doesn't exist
35try:
36 text = "hello"
37 result = text.nonexistent_method()
38except AttributeError as e:
39 print(f"AttributeError: {e}")Exception Hierarchy Understanding:
1# BaseException (top-level)
2# +-- SystemExit
3# +-- KeyboardInterrupt
4# +-- GeneratorExit
5# +-- Exception
6# +-- StopIteration
7# +-- ArithmeticError
8# | +-- ZeroDivisionError
9# | +-- OverflowError
10# +-- LookupError
11# | +-- IndexError
12# | +-- KeyError
13# +-- ValueError
14# +-- TypeError
15# +-- OSError
16# +-- FileNotFoundError
17# +-- PermissionError
18
19def demonstrate_exception_hierarchy():
20 """Show how exception inheritance works"""
21
22 try:
23 # This could raise various exceptions
24 result = risky_operation()
25 except LookupError as e:
26 # Catches both IndexError and KeyError
27 print(f"Lookup error occurred: {e}")
28 except ArithmeticError as e:
29 # Catches ZeroDivisionError, OverflowError, etc.
30 print(f"Math error occurred: {e}")
31 except Exception as e:
32 # Catches all other exceptions (except system-level)
33 print(f"General error occurred: {e}")Custom Exception Classes:
1class ValidationError(Exception):
2 """Custom exception for validation failures"""
3 def __init__(self, message, field=None, value=None):
4 super().__init__(message)
5 self.field = field
6 self.value = value
7 self.message = message
8
9 def __str__(self):
10 if self.field:
11 return f"Validation error in '{self.field}': {self.message} (value: {self.value})"
12 return self.message
13
14class DatabaseError(Exception):
15 """Custom exception for database operations"""
16 def __init__(self, message, error_code=None, query=None):
17 super().__init__(message)
18 self.error_code = error_code
19 self.query = query
20
21class UserNotFoundError(Exception):
22 """Specific exception for user-related operations"""
23 pass
24
25# Using custom exceptions
26def validate_user_age(age):
27 if not isinstance(age, int):
28 raise ValidationError("Age must be an integer", field="age", value=age)
29
30 if age < 0:
31 raise ValidationError("Age cannot be negative", field="age", value=age)
32
33 if age > 150:
34 raise ValidationError("Age seems unrealistic", field="age", value=age)
35
36 return True
37
38def find_user(user_id):
39 users = {1: "Alice", 2: "Bob", 3: "Carol"}
40
41 if user_id not in users:
42 raise UserNotFoundError(f"User with ID {user_id} not found")
43
44 return users[user_id]
45
46# Usage examples
47try:
48 validate_user_age("25") # Wrong type
49except ValidationError as e:
50 print(f"Validation failed: {e}")
51
52try:
53 user = find_user(999) # Non-existent user
54except UserNotFoundError as e:
55 print(f"User error: {e}")Exception Information and Traceback:
1import traceback
2import sys
3
4def get_exception_details():
5 """Demonstrate getting detailed exception information"""
6 try:
7 # Simulate a complex call stack
8 def level_3():
9 return 10 / 0
10
11 def level_2():
12 return level_3() * 2
13
14 def level_1():
15 return level_2() + 5
16
17 result = level_1()
18
19 except Exception as e:
20 # Get exception type, value, and traceback
21 exc_type, exc_value, exc_traceback = sys.exc_info()
22
23 print(f"Exception type: {exc_type.__name__}")
24 print(f"Exception value: {exc_value}")
25 print(f"Exception args: {e.args}")
26
27 # Print formatted traceback
28 print("\nFormatted traceback:")
29 traceback.print_exc()
30
31 # Get traceback as string
32 tb_string = traceback.format_exc()
33 print(f"\nTraceback as string:\n{tb_string}")
34
35 # Get specific traceback information
36 print("\nDetailed traceback info:")
37 for frame_info in traceback.extract_tb(exc_traceback):
38 print(f" File: {frame_info.filename}")
39 print(f" Line: {frame_info.lineno}")
40 print(f" Function: {frame_info.name}")
41 print(f" Code: {frame_info.line}")
42 print()
43
44get_exception_details()Best Practices for Exception Handling:
Professional exception handling goes beyond basic try-except blocks to include sophisticated error recovery, logging, and user experience considerations.
Comprehensive Try-Except-Finally-Else Pattern:
1import logging
2from typing import Optional, Dict, Any
3
4def process_user_data(user_data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
5 """
6 Process user data with comprehensive error handling
7 """
8 processed_data = None
9 temp_files = []
10
11 try:
12 # Validation phase
13 if not user_data:
14 raise ValueError("User data cannot be empty")
15
16 required_fields = ["name", "email", "age"]
17 missing_fields = [field for field in required_fields if field not in user_data]
18 if missing_fields:
19 raise ValidationError(f"Missing required fields: {missing_fields}")
20
21 # Processing phase
22 processed_data = {
23 "name": user_data["name"].strip().title(),
24 "email": user_data["email"].lower().strip(),
25 "age": int(user_data["age"]),
26 "processed_at": datetime.now().isoformat()
27 }
28
29 # Validate processed data
30 if processed_data["age"] < 0 or processed_data["age"] > 150:
31 raise ValidationError("Age must be between 0 and 150")
32
33 if "@" not in processed_data["email"]:
34 raise ValidationError("Invalid email format")
35
36 # Create temporary backup
37 temp_file = f"backup_{uuid.uuid4().hex}.json"
38 temp_files.append(temp_file)
39
40 with open(temp_file, 'w') as f:
41 json.dump(processed_data, f)
42
43 logging.info(f"Successfully processed user data for {processed_data['name']}")
44
45 except ValidationError as e:
46 logging.warning(f"Validation error: {e}")
47 raise # Re-raise validation errors
48
49 except ValueError as e:
50 logging.error(f"Value error in user data processing: {e}")
51 return None
52
53 except (IOError, OSError) as e:
54 logging.error(f"File system error: {e}")
55 # Continue processing despite backup failure
56
57 except Exception as e:
58 logging.error(f"Unexpected error processing user data: {e}")
59 logging.error(f"User data: {user_data}")
60 raise # Re-raise unexpected errors
61
62 else:
63 # Executed only if no exceptions occurred
64 logging.info("User data processing completed successfully")
65
66 finally:
67 # Cleanup temporary files
68 for temp_file in temp_files:
69 try:
70 if os.path.exists(temp_file):
71 os.remove(temp_file)
72 logging.debug(f"Cleaned up temporary file: {temp_file}")
73 except OSError as e:
74 logging.warning(f"Failed to clean up {temp_file}: {e}")
75
76 return processed_dataException Chaining and Context:
1class DataProcessingError(Exception):
2 """Exception for data processing failures"""
3 pass
4
5def process_data_with_context():
6 """Demonstrate exception chaining for better error context"""
7
8 try:
9 # Original operation that might fail
10 data = load_data_from_api()
11
12 except requests.RequestException as original_error:
13 # Chain the exception to preserve context
14 raise DataProcessingError(
15 "Failed to load data from API"
16 ) from original_error
17
18 except json.JSONDecodeError as original_error:
19 # Another chained exception
20 raise DataProcessingError(
21 "Invalid JSON response from API"
22 ) from original_error
23
24def handle_chained_exceptions():
25 """Handle exceptions with chained context"""
26 try:
27 process_data_with_context()
28
29 except DataProcessingError as e:
30 print(f"Processing error: {e}")
31
32 # Access the original exception
33 if e.__cause__:
34 print(f"Original error: {e.__cause__}")
35 print(f"Original error type: {type(e.__cause__).__name__}")
36
37 # Full traceback includes both exceptions
38 traceback.print_exc()Retry Mechanisms and Circuit Breakers:
1import time
2import random
3from functools import wraps
4from typing import Callable, Any
5
6def retry(max_attempts: int = 3, delay: float = 1.0, backoff: float = 2.0):
7 """Decorator for automatic retry with exponential backoff"""
8
9 def decorator(func: Callable) -> Callable:
10 @wraps(func)
11 def wrapper(*args, **kwargs) -> Any:
12 attempts = 0
13 current_delay = delay
14
15 while attempts < max_attempts:
16 try:
17 return func(*args, **kwargs)
18
19 except (requests.RequestException, ConnectionError) as e:
20 attempts += 1
21
22 if attempts >= max_attempts:
23 logging.error(f"Function {func.__name__} failed after {max_attempts} attempts")
24 raise
25
26 logging.warning(f"Attempt {attempts} failed for {func.__name__}: {e}")
27 logging.info(f"Retrying in {current_delay} seconds...")
28
29 time.sleep(current_delay)
30 current_delay *= backoff # Exponential backoff
31
32 except Exception as e:
33 # Don't retry for non-network errors
34 logging.error(f"Non-retryable error in {func.__name__}: {e}")
35 raise
36
37 return wrapper
38 return decorator
39
40class CircuitBreaker:
41 """Circuit breaker pattern for handling repeated failures"""
42
43 def __init__(self, failure_threshold: int = 5, timeout: float = 60.0):
44 self.failure_threshold = failure_threshold
45 self.timeout = timeout
46 self.failure_count = 0
47 self.last_failure_time = None
48 self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
49
50 def __call__(self, func: Callable) -> Callable:
51 @wraps(func)
52 def wrapper(*args, **kwargs) -> Any:
53 if self.state == "OPEN":
54 if time.time() - self.last_failure_time < self.timeout:
55 raise Exception("Circuit breaker is OPEN")
56 else:
57 self.state = "HALF_OPEN"
58
59 try:
60 result = func(*args, **kwargs)
61
62 if self.state == "HALF_OPEN":
63 self.state = "CLOSED"
64 self.failure_count = 0
65 logging.info("Circuit breaker reset to CLOSED")
66
67 return result
68
69 except Exception as e:
70 self.failure_count += 1
71 self.last_failure_time = time.time()
72
73 if self.failure_count >= self.failure_threshold:
74 self.state = "OPEN"
75 logging.warning(f"Circuit breaker opened after {self.failure_count} failures")
76
77 raise
78
79 return wrapper
80
81# Usage examples
82@retry(max_attempts=3, delay=1.0, backoff=2.0)
83@CircuitBreaker(failure_threshold=3, timeout=30.0)
84def unreliable_api_call():
85 """Simulate an unreliable API call"""
86 if random.random() < 0.7: # 70% failure rate
87 raise requests.RequestException("API temporarily unavailable")
88
89 return {"status": "success", "data": "Important data"}
90
91# Test the resilient function
92for i in range(10):
93 try:
94 result = unreliable_api_call()
95 print(f"Call {i+1}: {result}")
96 except Exception as e:
97 print(f"Call {i+1} failed: {e}")
98
99 time.sleep(1)Context Managers for Resource Management:
1from contextlib import contextmanager
2import sqlite3
3
4class DatabaseConnection:
5 """Database context manager with exception handling"""
6
7 def __init__(self, database_path: str):
8 self.database_path = database_path
9 self.connection = None
10 self.transaction_active = False
11
12 def __enter__(self):
13 try:
14 self.connection = sqlite3.connect(self.database_path)
15 self.connection.execute("BEGIN")
16 self.transaction_active = True
17 logging.info(f"Database connection opened: {self.database_path}")
18 return self.connection
19
20 except sqlite3.Error as e:
21 logging.error(f"Failed to connect to database: {e}")
22 raise
23
24 def __exit__(self, exc_type, exc_value, traceback):
25 if self.connection:
26 try:
27 if exc_type is None and self.transaction_active:
28 # No exception occurred, commit transaction
29 self.connection.execute("COMMIT")
30 logging.info("Database transaction committed")
31 else:
32 # Exception occurred, rollback transaction
33 self.connection.execute("ROLLBACK")
34 logging.warning("Database transaction rolled back")
35
36 except sqlite3.Error as e:
37 logging.error(f"Error during transaction cleanup: {e}")
38
39 finally:
40 self.connection.close()
41 logging.info("Database connection closed")
42
43 # Return False to propagate exceptions
44 return False
45
46@contextmanager
47def temporary_file_handler(filename: str):
48 """Context manager for temporary file with cleanup"""
49 temp_file = None
50
51 try:
52 temp_file = open(filename, 'w')
53 logging.info(f"Temporary file created: {filename}")
54 yield temp_file
55
56 except IOError as e:
57 logging.error(f"Error with temporary file {filename}: {e}")
58 raise
59
60 finally:
61 if temp_file:
62 temp_file.close()
63
64 try:
65 if os.path.exists(filename):
66 os.remove(filename)
67 logging.info(f"Temporary file cleaned up: {filename}")
68 except OSError as e:
69 logging.warning(f"Failed to clean up temporary file: {e}")
70
71# Usage examples
72try:
73 with DatabaseConnection("app.db") as db:
74 cursor = db.cursor()
75 cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)",
76 ("Alice", "[email protected]"))
77
78 # If an exception occurs here, transaction will be rolled back
79 cursor.execute("SELECT * FROM users")
80
81except sqlite3.Error as e:
82 logging.error(f"Database operation failed: {e}")
83
84with temporary_file_handler("temp_data.txt") as temp_file:
85 temp_file.write("Temporary data")
86 # File will be automatically cleaned upAdvanced Exception Handling Techniques:
Effective debugging is essential for developing and maintaining Python applications. Python provides powerful built-in debugging tools and techniques for identifying and fixing issues.
Python Debugger (pdb) - The Built-in Debugger:
1import pdb
2
3def calculate_statistics(numbers):
4 """Calculate basic statistics with debugging"""
5
6 # Set a breakpoint
7 pdb.set_trace()
8
9 if not numbers:
10 return None
11
12 total = sum(numbers)
13 count = len(numbers)
14 average = total / count
15
16 # Another debugging point
17 pdb.set_trace()
18
19 # Find min and max
20 minimum = min(numbers)
21 maximum = max(numbers)
22
23 return {
24 'count': count,
25 'sum': total,
26 'average': average,
27 'min': minimum,
28 'max': maximum
29 }
30
31# Usage with pdb commands:
32# (Pdb) l - list current code
33# (Pdb) n - next line
34# (Pdb) s - step into function
35# (Pdb) c - continue
36# (Pdb) p variable_name - print variable
37# (Pdb) pp variable_name - pretty print
38# (Pdb) w - where (stack trace)
39# (Pdb) u - up in stack
40# (Pdb) d - down in stack
41# (Pdb) h - help
42
43# Alternative: post-mortem debugging
44try:
45 result = calculate_statistics([])
46except Exception:
47 pdb.post_mortem() # Debug at point of exceptionAdvanced Debugging with Custom Debug Functions:
1import inspect
2import functools
3from typing import Any, Callable
4
5def debug_info(func: Callable) -> Callable:
6 """Decorator to add debug information to functions"""
7
8 @functools.wraps(func)
9 def wrapper(*args, **kwargs) -> Any:
10 # Get caller information
11 frame = inspect.currentframe().f_back
12 caller_info = inspect.getframeinfo(frame)
13
14 print(f"\n=== DEBUG: {func.__name__} ===")
15 print(f"Called from: {caller_info.filename}:{caller_info.lineno}")
16 print(f"Function: {caller_info.function}")
17 print(f"Arguments: args={args}, kwargs={kwargs}")
18
19 # Execute function and measure time
20 import time
21 start_time = time.time()
22
23 try:
24 result = func(*args, **kwargs)
25 execution_time = time.time() - start_time
26
27 print(f"Result: {result}")
28 print(f"Execution time: {execution_time:.4f} seconds")
29 print(f"=== END DEBUG: {func.__name__} ===\n")
30
31 return result
32
33 except Exception as e:
34 execution_time = time.time() - start_time
35
36 print(f"Exception: {type(e).__name__}: {e}")
37 print(f"Execution time: {execution_time:.4f} seconds")
38 print(f"=== END DEBUG (ERROR): {func.__name__} ===\n")
39
40 raise
41
42 return wrapper
43
44class DebugContext:
45 """Context manager for detailed debugging"""
46
47 def __init__(self, name: str, log_variables: bool = True):
48 self.name = name
49 self.log_variables = log_variables
50 self.start_time = None
51
52 def __enter__(self):
53 self.start_time = time.time()
54 frame = inspect.currentframe().f_back
55
56 print(f"\n>>> Entering debug context: {self.name}")
57
58 if self.log_variables:
59 local_vars = frame.f_locals
60 print("Local variables:")
61 for name, value in local_vars.items():
62 if not name.startswith('_'):
63 print(f" {name} = {value} ({type(value).__name__})")
64
65 return self
66
67 def __exit__(self, exc_type, exc_value, traceback):
68 execution_time = time.time() - self.start_time
69
70 if exc_type:
71 print(f"<<< Exiting debug context: {self.name} (ERROR)")
72 print(f"Exception: {exc_type.__name__}: {exc_value}")
73 else:
74 print(f"<<< Exiting debug context: {self.name} (SUCCESS)")
75
76 print(f"Context execution time: {execution_time:.4f} seconds\n")
77
78 return False # Don't suppress exceptions
79
80# Usage examples
81@debug_info
82def complex_calculation(x, y, operation="add"):
83 """A function with debug information"""
84
85 with DebugContext(f"calculation_{operation}"):
86 if operation == "add":
87 result = x + y
88 elif operation == "multiply":
89 result = x * y
90 elif operation == "divide":
91 if y == 0:
92 raise ValueError("Cannot divide by zero")
93 result = x / y
94 else:
95 raise ValueError(f"Unknown operation: {operation}")
96
97 return result
98
99# Test the debugging
100try:
101 result1 = complex_calculation(10, 5, "add")
102 result2 = complex_calculation(10, 0, "divide") # Will raise exception
103except Exception as e:
104 print(f"Caught exception: {e}")Logging for Debugging and Monitoring:
1import logging
2import sys
3from datetime import datetime
4from pathlib import Path
5
6class AdvancedLogger:
7 """Advanced logging setup for debugging and production"""
8
9 def __init__(self, name: str, log_file: str = None, level: str = "INFO"):
10 self.logger = logging.getLogger(name)
11 self.logger.setLevel(getattr(logging, level.upper()))
12
13 # Clear existing handlers
14 self.logger.handlers.clear()
15
16 # Create formatters
17 detailed_formatter = logging.Formatter(
18 '%(asctime)s | %(name)s | %(levelname)s | %(filename)s:%(lineno)d | %(funcName)s | %(message)s'
19 )
20
21 simple_formatter = logging.Formatter(
22 '%(asctime)s | %(levelname)s | %(message)s'
23 )
24
25 # Console handler
26 console_handler = logging.StreamHandler(sys.stdout)
27 console_handler.setLevel(logging.INFO)
28 console_handler.setFormatter(simple_formatter)
29 self.logger.addHandler(console_handler)
30
31 # File handler (if specified)
32 if log_file:
33 log_path = Path(log_file)
34 log_path.parent.mkdir(parents=True, exist_ok=True)
35
36 file_handler = logging.FileHandler(log_file)
37 file_handler.setLevel(logging.DEBUG)
38 file_handler.setFormatter(detailed_formatter)
39 self.logger.addHandler(file_handler)
40
41 # Error file handler
42 if log_file:
43 error_file = log_path.parent / f"error_{log_path.name}"
44 error_handler = logging.FileHandler(error_file)
45 error_handler.setLevel(logging.ERROR)
46 error_handler.setFormatter(detailed_formatter)
47 self.logger.addHandler(error_handler)
48
49 def debug(self, message: str, **kwargs):
50 self.logger.debug(message, extra=kwargs)
51
52 def info(self, message: str, **kwargs):
53 self.logger.info(message, extra=kwargs)
54
55 def warning(self, message: str, **kwargs):
56 self.logger.warning(message, extra=kwargs)
57
58 def error(self, message: str, exception: Exception = None, **kwargs):
59 if exception:
60 self.logger.error(f"{message} | Exception: {exception}", extra=kwargs)
61 else:
62 self.logger.error(message, extra=kwargs)
63
64 def critical(self, message: str, **kwargs):
65 self.logger.critical(message, extra=kwargs)
66
67def log_function_calls(logger: AdvancedLogger):
68 """Decorator to log function entry and exit"""
69
70 def decorator(func: Callable) -> Callable:
71 @functools.wraps(func)
72 def wrapper(*args, **kwargs) -> Any:
73 # Log function entry
74 logger.debug(f"Entering {func.__name__} with args={args}, kwargs={kwargs}")
75
76 start_time = time.time()
77
78 try:
79 result = func(*args, **kwargs)
80 execution_time = time.time() - start_time
81
82 logger.debug(f"Exiting {func.__name__} successfully in {execution_time:.4f}s")
83 return result
84
85 except Exception as e:
86 execution_time = time.time() - start_time
87 logger.error(f"Exception in {func.__name__} after {execution_time:.4f}s", exception=e)
88 raise
89
90 return wrapper
91 return decorator
92
93# Setup logging
94app_logger = AdvancedLogger("MyApp", "logs/app.log", "DEBUG")
95
96@log_function_calls(app_logger)
97def process_user_registration(username: str, email: str, age: int):
98 """Process user registration with comprehensive logging"""
99
100 app_logger.info(f"Processing registration for user: {username}")
101
102 # Validation
103 if not username or len(username) < 3:
104 app_logger.warning(f"Invalid username: {username}")
105 raise ValueError("Username must be at least 3 characters")
106
107 if "@" not in email:
108 app_logger.warning(f"Invalid email format: {email}")
109 raise ValueError("Invalid email format")
110
111 if age < 18:
112 app_logger.warning(f"Underage user registration attempt: {age}")
113 raise ValueError("User must be at least 18 years old")
114
115 # Simulate processing
116 app_logger.debug("Validating email uniqueness")
117 # ... email validation logic ...
118
119 app_logger.debug("Creating user account")
120 # ... account creation logic ...
121
122 app_logger.info(f"Successfully registered user: {username}")
123
124 return {"user_id": 12345, "username": username, "status": "active"}
125
126# Test with logging
127try:
128 user = process_user_registration("alice_smith", "[email protected]", 25)
129 app_logger.info("Registration completed successfully")
130except Exception as e:
131 app_logger.error("Registration failed", exception=e)Performance Profiling and Memory Debugging:
1import cProfile
2import pstats
3import tracemalloc
4from memory_profiler import profile
5import psutil
6import os
7
8def profile_function(func: Callable) -> Callable:
9 """Decorator for performance profiling"""
10
11 @functools.wraps(func)
12 def wrapper(*args, **kwargs) -> Any:
13 # Start profiling
14 profiler = cProfile.Profile()
15 profiler.enable()
16
17 # Start memory tracking
18 tracemalloc.start()
19 process = psutil.Process(os.getpid())
20 start_memory = process.memory_info().rss / 1024 / 1024 # MB
21
22 try:
23 result = func(*args, **kwargs)
24 return result
25
26 finally:
27 # Stop profiling
28 profiler.disable()
29
30 # Get memory stats
31 current, peak = tracemalloc.get_traced_memory()
32 tracemalloc.stop()
33 end_memory = process.memory_info().rss / 1024 / 1024 # MB
34
35 # Print performance stats
36 print(f"\n=== Performance Profile: {func.__name__} ===")
37 print(f"Memory usage: {current / 1024 / 1024:.2f} MB current, {peak / 1024 / 1024:.2f} MB peak")
38 print(f"Process memory: {start_memory:.2f} MB -> {end_memory:.2f} MB ({end_memory - start_memory:+.2f} MB)")
39
40 # Print top time-consuming functions
41 stats = pstats.Stats(profiler)
42 stats.sort_stats('cumulative')
43 stats.print_stats(10) # Top 10 functions
44
45 print(f"=== End Profile: {func.__name__} ===\n")
46
47 return wrapper
48
49@profile_function
50def memory_intensive_task():
51 """A task that uses significant memory and CPU"""
52
53 # Create large data structures
54 large_list = list(range(1000000))
55 large_dict = {i: f"value_{i}" for i in range(100000)}
56
57 # Perform computations
58 result = []
59 for i in range(10000):
60 result.append(sum(large_list[i:i+100]))
61
62 return len(result)
63
64# Run profiling
65result = memory_intensive_task()
66
67# Alternative: Line-by-line memory profiling (requires memory_profiler)
68@profile
69def line_by_line_memory_example():
70 """Function with line-by-line memory profiling"""
71 a = [1] * (256 * 1024) # Line 1
72 b = [2] * (512 * 1024) # Line 2
73 c = [3] * (1024 * 1024) # Line 3
74 del b # Line 4
75 return a + c # Line 5
76
77# Run with: python -m memory_profiler script.pyProfessional Debugging Practices:
Production applications require sophisticated error handling strategies that go beyond basic exception catching to include monitoring, alerting, and automated recovery.
Error Logging and Monitoring System:
1import logging
2import json
3import smtplib
4import time
5from email.mime.text import MIMEText
6from datetime import datetime, timedelta
7from collections import defaultdict, deque
8from typing import Dict, List, Optional, Any
9from dataclasses import dataclass
10from threading import Lock
11import sqlite3
12
13@dataclass
14class ErrorMetrics:
15 """Data class for error metrics"""
16 error_type: str
17 count: int
18 first_occurrence: datetime
19 last_occurrence: datetime
20 affected_users: set
21 severity: str = "medium"
22
23class ProductionErrorHandler:
24 """Comprehensive error handling system for production applications"""
25
26 def __init__(self, database_path: str = "error_logs.db"):
27 self.database_path = database_path
28 self.error_counts = defaultdict(int)
29 self.error_history = deque(maxlen=1000) # Keep last 1000 errors
30 self.error_metrics = {}
31 self.lock = Lock()
32
33 # Setup database
34 self._setup_database()
35
36 # Setup logging
37 self._setup_logging()
38
39 # Error thresholds
40 self.error_thresholds = {
41 "critical": 1, # Alert on first critical error
42 "high": 5, # Alert after 5 high-severity errors
43 "medium": 20, # Alert after 20 medium-severity errors
44 "low": 100 # Alert after 100 low-severity errors
45 }
46
47 def _setup_database(self):
48 """Initialize error logging database"""
49 with sqlite3.connect(self.database_path) as conn:
50 conn.execute("""
51 CREATE TABLE IF NOT EXISTS error_logs (
52 id INTEGER PRIMARY KEY AUTOINCREMENT,
53 timestamp TEXT NOT NULL,
54 error_type TEXT NOT NULL,
55 error_message TEXT NOT NULL,
56 severity TEXT NOT NULL,
57 user_id TEXT,
58 request_id TEXT,
59 stack_trace TEXT,
60 context_data TEXT,
61 resolved BOOLEAN DEFAULT FALSE
62 )
63 """)
64
65 conn.execute("""
66 CREATE INDEX IF NOT EXISTS idx_error_type ON error_logs(error_type)
67 """)
68
69 conn.execute("""
70 CREATE INDEX IF NOT EXISTS idx_timestamp ON error_logs(timestamp)
71 """)
72
73 def _setup_logging(self):
74 """Setup structured logging"""
75 self.logger = logging.getLogger("ProductionErrorHandler")
76 self.logger.setLevel(logging.DEBUG)
77
78 # JSON formatter for structured logging
79 formatter = logging.Formatter('%(message)s')
80
81 # File handler for error logs
82 file_handler = logging.FileHandler("production_errors.json")
83 file_handler.setFormatter(formatter)
84 self.logger.addHandler(file_handler)
85
86 def log_error(self,
87 error: Exception,
88 severity: str = "medium",
89 user_id: str = None,
90 request_id: str = None,
91 context: Dict[str, Any] = None) -> str:
92 """Log an error with comprehensive details"""
93
94 error_id = f"ERR_{int(time.time())}_{hash(str(error)) % 10000:04d}"
95 timestamp = datetime.now()
96
97 # Get stack trace
98 import traceback
99 stack_trace = traceback.format_exc()
100
101 # Create error log entry
102 error_log = {
103 "error_id": error_id,
104 "timestamp": timestamp.isoformat(),
105 "error_type": type(error).__name__,
106 "error_message": str(error),
107 "severity": severity,
108 "user_id": user_id,
109 "request_id": request_id,
110 "stack_trace": stack_trace,
111 "context": context or {},
112 "hostname": os.getenv("HOSTNAME", "unknown"),
113 "environment": os.getenv("ENVIRONMENT", "production")
114 }
115
116 # Log to JSON file
117 self.logger.error(json.dumps(error_log))
118
119 # Store in database
120 self._store_error_in_database(error_log)
121
122 # Update metrics
123 self._update_error_metrics(error_log)
124
125 # Check if alert should be sent
126 self._check_alert_thresholds(error_log)
127
128 return error_id
129
130 def _store_error_in_database(self, error_log: Dict[str, Any]):
131 """Store error in database"""
132 try:
133 with sqlite3.connect(self.database_path) as conn:
134 conn.execute("""
135 INSERT INTO error_logs
136 (timestamp, error_type, error_message, severity, user_id,
137 request_id, stack_trace, context_data)
138 VALUES (?, ?, ?, ?, ?, ?, ?, ?)
139 """, (
140 error_log["timestamp"],
141 error_log["error_type"],
142 error_log["error_message"],
143 error_log["severity"],
144 error_log["user_id"],
145 error_log["request_id"],
146 error_log["stack_trace"],
147 json.dumps(error_log["context"])
148 ))
149 except Exception as e:
150 # Don't let logging errors break the application
151 print(f"Failed to store error in database: {e}")
152
153 def _update_error_metrics(self, error_log: Dict[str, Any]):
154 """Update error metrics and statistics"""
155 with self.lock:
156 error_type = error_log["error_type"]
157 timestamp = datetime.fromisoformat(error_log["timestamp"])
158 user_id = error_log["user_id"]
159
160 if error_type not in self.error_metrics:
161 self.error_metrics[error_type] = ErrorMetrics(
162 error_type=error_type,
163 count=0,
164 first_occurrence=timestamp,
165 last_occurrence=timestamp,
166 affected_users=set(),
167 severity=error_log["severity"]
168 )
169
170 metrics = self.error_metrics[error_type]
171 metrics.count += 1
172 metrics.last_occurrence = timestamp
173
174 if user_id:
175 metrics.affected_users.add(user_id)
176
177 # Update severity if current error is more severe
178 severity_order = {"low": 1, "medium": 2, "high": 3, "critical": 4}
179 if severity_order[error_log["severity"]] > severity_order[metrics.severity]:
180 metrics.severity = error_log["severity"]
181
182 # Add to recent history
183 self.error_history.append(error_log)
184
185 def _check_alert_thresholds(self, error_log: Dict[str, Any]):
186 """Check if error thresholds have been exceeded"""
187 error_type = error_log["error_type"]
188 severity = error_log["severity"]
189
190 if error_type in self.error_metrics:
191 count = self.error_metrics[error_type].count
192 threshold = self.error_thresholds.get(severity, float('inf'))
193
194 if count >= threshold and count % threshold == 0:
195 self._send_alert(error_log, count)
196
197 def _send_alert(self, error_log: Dict[str, Any], count: int):
198 """Send alert for critical errors (placeholder implementation)"""
199 alert_message = f"""
200 PRODUCTION ERROR ALERT
201
202 Error Type: {error_log['error_type']}
203 Severity: {error_log['severity']}
204 Count: {count} occurrences
205 Message: {error_log['error_message']}
206 User ID: {error_log['user_id']}
207 Request ID: {error_log['request_id']}
208 Timestamp: {error_log['timestamp']}
209
210 Stack Trace:
211 {error_log['stack_trace']}
212
213 Context:
214 {json.dumps(error_log['context'], indent=2)}
215 """
216
217 print(f"🚨 ALERT SENT: {error_log['error_type']} ({count} occurrences)")
218
219 # In production, this would integrate with:
220 # - Email/SMS alerting
221 # - Slack/Teams notifications
222 # - PagerDuty/OpsGenie
223 # - Monitoring dashboards (Grafana, DataDog, etc.)
224
225 def get_error_statistics(self, hours: int = 24) -> Dict[str, Any]:
226 """Get error statistics for the specified time period"""
227 cutoff_time = datetime.now() - timedelta(hours=hours)
228
229 with sqlite3.connect(self.database_path) as conn:
230 cursor = conn.cursor()
231
232 # Get error counts by type
233 cursor.execute("""
234 SELECT error_type, severity, COUNT(*) as count
235 FROM error_logs
236 WHERE timestamp > ?
237 GROUP BY error_type, severity
238 ORDER BY count DESC
239 """, (cutoff_time.isoformat(),))
240
241 error_counts = cursor.fetchall()
242
243 # Get hourly error trend
244 cursor.execute("""
245 SELECT
246 strftime('%Y-%m-%d %H:00:00', timestamp) as hour,
247 COUNT(*) as count
248 FROM error_logs
249 WHERE timestamp > ?
250 GROUP BY hour
251 ORDER BY hour
252 """, (cutoff_time.isoformat(),))
253
254 hourly_trend = cursor.fetchall()
255
256 # Get most affected users
257 cursor.execute("""
258 SELECT user_id, COUNT(*) as error_count
259 FROM error_logs
260 WHERE timestamp > ? AND user_id IS NOT NULL
261 GROUP BY user_id
262 ORDER BY error_count DESC
263 LIMIT 10
264 """, (cutoff_time.isoformat(),))
265
266 affected_users = cursor.fetchall()
267
268 return {
269 "time_period": f"Last {hours} hours",
270 "error_counts": error_counts,
271 "hourly_trend": hourly_trend,
272 "most_affected_users": affected_users,
273 "total_errors": sum(count for _, _, count in error_counts),
274 "unique_error_types": len(error_counts)
275 }
276
277 def generate_error_report(self) -> str:
278 """Generate comprehensive error report"""
279 stats = self.get_error_statistics(24)
280
281 report = f"""
282 PRODUCTION ERROR REPORT
283 {'=' * 50}
284
285 Time Period: {stats['time_period']}
286 Total Errors: {stats['total_errors']}
287 Unique Error Types: {stats['unique_error_types']}
288
289 TOP ERROR TYPES:
290 {'-' * 30}
291 """
292
293 for error_type, severity, count in stats['error_counts'][:10]:
294 report += f"\n{error_type} ({severity}): {count} occurrences"
295
296 if stats['most_affected_users']:
297 report += f"\n\nMOST AFFECTED USERS:\n{'-' * 30}"
298 for user_id, error_count in stats['most_affected_users'][:5]:
299 report += f"\nUser {user_id}: {error_count} errors"
300
301 return report
302
303# Global error handler instance
304error_handler = ProductionErrorHandler()
305
306def production_exception_handler(func: Callable) -> Callable:
307 """Decorator for production exception handling"""
308
309 @functools.wraps(func)
310 def wrapper(*args, **kwargs) -> Any:
311 try:
312 return func(*args, **kwargs)
313
314 except ValueError as e:
315 error_id = error_handler.log_error(
316 e,
317 severity="medium",
318 context={"function": func.__name__, "args": str(args)[:100]}
319 )
320 # Return user-friendly error
321 return {"error": "Invalid input provided", "error_id": error_id}
322
323 except ConnectionError as e:
324 error_id = error_handler.log_error(
325 e,
326 severity="high",
327 context={"function": func.__name__, "service": "external"}
328 )
329 # Return user-friendly error
330 return {"error": "Service temporarily unavailable", "error_id": error_id}
331
332 except Exception as e:
333 error_id = error_handler.log_error(
334 e,
335 severity="critical",
336 context={"function": func.__name__, "unexpected": True}
337 )
338 # Return generic error for unexpected exceptions
339 return {"error": "An unexpected error occurred", "error_id": error_id}
340
341 return wrapper
342
343# Usage example
344@production_exception_handler
345def process_payment(user_id: str, amount: float, payment_method: str):
346 """Process payment with production error handling"""
347
348 if amount <= 0:
349 raise ValueError("Payment amount must be positive")
350
351 if payment_method not in ["credit_card", "paypal", "bank_transfer"]:
352 raise ValueError("Invalid payment method")
353
354 # Simulate external service call
355 if payment_method == "credit_card" and amount > 1000:
356 raise ConnectionError("Payment gateway timeout")
357
358 # Simulate unexpected error
359 if user_id == "test_error":
360 raise RuntimeError("Unexpected system error")
361
362 return {"status": "success", "transaction_id": f"TXN_{int(time.time())}"}
363
364# Test error handling
365results = []
366test_cases = [
367 ("user123", 50.0, "credit_card"), # Success
368 ("user456", -10.0, "credit_card"), # ValueError
369 ("user789", 1500.0, "credit_card"), # ConnectionError
370 ("test_error", 100.0, "paypal"), # Unexpected error
371]
372
373for user_id, amount, method in test_cases:
374 result = process_payment(user_id, amount, method)
375 results.append(result)
376 print(f"Payment result: {result}")
377
378# Generate error report
379print("\n" + error_handler.generate_error_report())Health Checks and Circuit Breakers:
1class HealthMonitor:
2 """System health monitoring for production applications"""
3
4 def __init__(self):
5 self.health_checks = {}
6 self.circuit_breakers = {}
7
8 def register_health_check(self, name: str, check_function: Callable) -> None:
9 """Register a health check function"""
10 self.health_checks[name] = check_function
11
12 def run_health_checks(self) -> Dict[str, Any]:
13 """Run all health checks and return status"""
14 results = {}
15 overall_healthy = True
16
17 for name, check_func in self.health_checks.items():
18 try:
19 start_time = time.time()
20 result = check_func()
21 response_time = time.time() - start_time
22
23 results[name] = {
24 "status": "healthy" if result else "unhealthy",
25 "response_time": response_time,
26 "details": result if isinstance(result, dict) else {}
27 }
28
29 if not result:
30 overall_healthy = False
31
32 except Exception as e:
33 results[name] = {
34 "status": "error",
35 "error": str(e),
36 "response_time": None
37 }
38 overall_healthy = False
39
40 return {
41 "overall_status": "healthy" if overall_healthy else "unhealthy",
42 "timestamp": datetime.now().isoformat(),
43 "checks": results
44 }
45
46# Health check functions
47def database_health_check():
48 """Check database connectivity"""
49 try:
50 with sqlite3.connect("app.db", timeout=5) as conn:
51 conn.execute("SELECT 1")
52 return True
53 except Exception:
54 return False
55
56def external_api_health_check():
57 """Check external API connectivity"""
58 try:
59 # Simulate API check
60 import requests
61 response = requests.get("https://api.example.com/health", timeout=5)
62 return response.status_code == 200
63 except Exception:
64 return False
65
66def memory_health_check():
67 """Check system memory usage"""
68 try:
69 process = psutil.Process()
70 memory_percent = process.memory_percent()
71 return {
72 "healthy": memory_percent < 80,
73 "memory_usage_percent": memory_percent
74 }
75 except Exception:
76 return False
77
78# Setup health monitoring
79health_monitor = HealthMonitor()
80health_monitor.register_health_check("database", database_health_check)
81health_monitor.register_health_check("external_api", external_api_health_check)
82health_monitor.register_health_check("memory", memory_health_check)
83
84# Run health checks
85health_status = health_monitor.run_health_checks()
86print(json.dumps(health_status, indent=2))Production Error Handling Best Practices:
Build a production-ready web application demonstrating comprehensive exception handling and debugging techniques
1#!/usr/bin/env python3
2"""
3Robust Web Application with Complete Error Handling
4Demonstrates production-ready exception handling, debugging, and monitoring
5"""
6
7import os
8import json
9import time
10import sqlite3
11import logging
12import traceback
13import functools
14from datetime import datetime, timedelta
15from typing import Dict, List, Optional, Any, Callable
16from dataclasses import dataclass, asdict
17from contextlib import contextmanager
18from pathlib import Path
19
20# Custom Exception Classes
21class ValidationError(Exception):
22 """Custom exception for validation failures"""
23 def __init__(self, message: str, field: str = None, value: Any = None):
24 super().__init__(message)
25 self.field = field
26 self.value = value
27 self.timestamp = datetime.now()
28
29class BusinessLogicError(Exception):
30 """Custom exception for business logic violations"""
31 def __init__(self, message: str, error_code: str = None):
32 super().__init__(message)
33 self.error_code = error_code
34 self.timestamp = datetime.now()
35
36class ExternalServiceError(Exception):
37 """Custom exception for external service failures"""
38 def __init__(self, message: str, service_name: str = None, status_code: int = None):
39 super().__init__(message)
40 self.service_name = service_name
41 self.status_code = status_code
42 self.timestamp = datetime.now()
43
44@dataclass
45class User:
46 """User data model"""
47 id: int
48 username: str
49 email: str
50 age: int
51 created_at: datetime
52 active: bool = True
53
54class DatabaseManager:
55 """Database manager with comprehensive error handling"""
56
57 def __init__(self, db_path: str = "app.db"):
58 self.db_path = db_path
59 self.setup_database()
60
61 def setup_database(self):
62 """Initialize database with error handling"""
63 try:
64 with sqlite3.connect(self.db_path) as conn:
65 conn.execute("""
66 CREATE TABLE IF NOT EXISTS users (
67 id INTEGER PRIMARY KEY AUTOINCREMENT,
68 username TEXT UNIQUE NOT NULL,
69 email TEXT UNIQUE NOT NULL,
70 age INTEGER NOT NULL,
71 created_at TEXT NOT NULL,
72 active BOOLEAN DEFAULT TRUE
73 )
74 """)
75
76 conn.execute("""
77 CREATE TABLE IF NOT EXISTS user_sessions (
78 id INTEGER PRIMARY KEY AUTOINCREMENT,
79 user_id INTEGER,
80 session_token TEXT UNIQUE NOT NULL,
81 created_at TEXT NOT NULL,
82 expires_at TEXT NOT NULL,
83 FOREIGN KEY (user_id) REFERENCES users (id)
84 )
85 """)
86
87 logging.info("Database initialized successfully")
88
89 except sqlite3.Error as e:
90 logging.error(f"Database initialization failed: {e}")
91 raise ExternalServiceError(f"Database setup failed: {e}", "database")
92
93 @contextmanager
94 def get_connection(self):
95 """Context manager for database connections"""
96 conn = None
97 try:
98 conn = sqlite3.connect(self.db_path, timeout=30)
99 conn.row_factory = sqlite3.Row # Enable dict-like access
100 yield conn
101
102 except sqlite3.Error as e:
103 if conn:
104 conn.rollback()
105 logging.error(f"Database error: {e}")
106 raise ExternalServiceError(f"Database operation failed: {e}", "database")
107
108 finally:
109 if conn:
110 conn.close()
111
112 def create_user(self, username: str, email: str, age: int) -> User:
113 """Create a new user with validation"""
114 # Validate input
115 if not username or len(username) < 3:
116 raise ValidationError("Username must be at least 3 characters", "username", username)
117
118 if not email or "@" not in email:
119 raise ValidationError("Invalid email format", "email", email)
120
121 if not isinstance(age, int) or age < 13 or age > 120:
122 raise ValidationError("Age must be between 13 and 120", "age", age)
123
124 try:
125 with self.get_connection() as conn:
126 cursor = conn.cursor()
127
128 # Check for existing username
129 cursor.execute("SELECT id FROM users WHERE username = ?", (username,))
130 if cursor.fetchone():
131 raise BusinessLogicError(f"Username '{username}' already exists", "DUPLICATE_USERNAME")
132
133 # Check for existing email
134 cursor.execute("SELECT id FROM users WHERE email = ?", (email,))
135 if cursor.fetchone():
136 raise BusinessLogicError(f"Email '{email}' already registered", "DUPLICATE_EMAIL")
137
138 # Create user
139 created_at = datetime.now().isoformat()
140 cursor.execute("""
141 INSERT INTO users (username, email, age, created_at)
142 VALUES (?, ?, ?, ?)
143 """, (username, email, age, created_at))
144
145 user_id = cursor.lastrowid
146 conn.commit()
147
148 user = User(
149 id=user_id,
150 username=username,
151 email=email,
152 age=age,
153 created_at=datetime.fromisoformat(created_at)
154 )
155
156 logging.info(f"User created successfully: {username} (ID: {user_id})")
157 return user
158
159 except (ValidationError, BusinessLogicError):
160 raise # Re-raise custom exceptions
161 except Exception as e:
162 logging.error(f"Unexpected error creating user: {e}")
163 raise ExternalServiceError(f"User creation failed: {e}", "database")
164
165 def get_user(self, user_id: int) -> Optional[User]:
166 """Get user by ID with error handling"""
167 try:
168 with self.get_connection() as conn:
169 cursor = conn.cursor()
170 cursor.execute("""
171 SELECT id, username, email, age, created_at, active
172 FROM users WHERE id = ?
173 """, (user_id,))
174
175 row = cursor.fetchone()
176 if not row:
177 return None
178
179 return User(
180 id=row["id"],
181 username=row["username"],
182 email=row["email"],
183 age=row["age"],
184 created_at=datetime.fromisoformat(row["created_at"]),
185 active=bool(row["active"])
186 )
187
188 except Exception as e:
189 logging.error(f"Error retrieving user {user_id}: {e}")
190 raise ExternalServiceError(f"User retrieval failed: {e}", "database")
191
192 def update_user(self, user_id: int, **updates) -> bool:
193 """Update user with validation"""
194 if not updates:
195 raise ValidationError("No updates provided")
196
197 # Validate updates
198 allowed_fields = {"username", "email", "age", "active"}
199 invalid_fields = set(updates.keys()) - allowed_fields
200 if invalid_fields:
201 raise ValidationError(f"Invalid fields for update: {invalid_fields}")
202
203 # Validate specific fields
204 if "username" in updates and (not updates["username"] or len(updates["username"]) < 3):
205 raise ValidationError("Username must be at least 3 characters", "username")
206
207 if "email" in updates and (not updates["email"] or "@" not in updates["email"]):
208 raise ValidationError("Invalid email format", "email")
209
210 if "age" in updates and (not isinstance(updates["age"], int) or updates["age"] < 13 or updates["age"] > 120):
211 raise ValidationError("Age must be between 13 and 120", "age")
212
213 try:
214 with self.get_connection() as conn:
215 cursor = conn.cursor()
216
217 # Check if user exists
218 cursor.execute("SELECT id FROM users WHERE id = ?", (user_id,))
219 if not cursor.fetchone():
220 raise BusinessLogicError(f"User with ID {user_id} not found", "USER_NOT_FOUND")
221
222 # Build dynamic update query
223 set_clauses = []
224 values = []
225
226 for field, value in updates.items():
227 set_clauses.append(f"{field} = ?")
228 values.append(value)
229
230 values.append(user_id) # For WHERE clause
231
232 query = f"UPDATE users SET {', '.join(set_clauses)} WHERE id = ?"
233 cursor.execute(query, values)
234
235 conn.commit()
236
237 logging.info(f"User {user_id} updated successfully: {updates}")
238 return True
239
240 except (ValidationError, BusinessLogicError):
241 raise
242 except Exception as e:
243 logging.error(f"Error updating user {user_id}: {e}")
244 raise ExternalServiceError(f"User update failed: {e}", "database")
245
246class UserService:
247 """User service with comprehensive error handling"""
248
249 def __init__(self, db_manager: DatabaseManager):
250 self.db = db_manager
251
252 def register_user(self, username: str, email: str, age: int) -> Dict[str, Any]:
253 """Register a new user with comprehensive error handling"""
254 try:
255 # Business logic validation
256 if self._is_email_blacklisted(email):
257 raise BusinessLogicError(f"Email domain is not allowed: {email}", "BLACKLISTED_EMAIL")
258
259 # Create user
260 user = self.db.create_user(username, email, age)
261
262 # Send welcome email (simulate)
263 self._send_welcome_email(user.email)
264
265 # Log successful registration
266 logging.info(f"User registration completed: {user.username}")
267
268 return {
269 "success": True,
270 "user_id": user.id,
271 "username": user.username,
272 "message": "User registered successfully"
273 }
274
275 except ValidationError as e:
276 logging.warning(f"User registration validation failed: {e}")
277 return {
278 "success": False,
279 "error_type": "validation_error",
280 "message": str(e),
281 "field": e.field,
282 "value": e.value
283 }
284
285 except BusinessLogicError as e:
286 logging.warning(f"User registration business logic error: {e}")
287 return {
288 "success": False,
289 "error_type": "business_error",
290 "message": str(e),
291 "error_code": e.error_code
292 }
293
294 except ExternalServiceError as e:
295 logging.error(f"External service error during registration: {e}")
296 return {
297 "success": False,
298 "error_type": "service_error",
299 "message": "Registration temporarily unavailable",
300 "service": e.service_name
301 }
302
303 except Exception as e:
304 logging.error(f"Unexpected error during user registration: {e}")
305 logging.error(f"Traceback: {traceback.format_exc()}")
306 return {
307 "success": False,
308 "error_type": "system_error",
309 "message": "An unexpected error occurred"
310 }
311
312 def _is_email_blacklisted(self, email: str) -> bool:
313 """Check if email domain is blacklisted"""
314 blacklisted_domains = {"tempmail.com", "10minutemail.com", "throwaway.email"}
315 domain = email.split("@")[-1].lower()
316 return domain in blacklisted_domains
317
318 def _send_welcome_email(self, email: str):
319 """Simulate sending welcome email"""
320 # Simulate external service call
321 import random
322
323 if random.random() < 0.1: # 10% failure rate
324 raise ExternalServiceError("Email service temporarily unavailable", "email_service", 503)
325
326 logging.info(f"Welcome email sent to: {email}")
327
328 def get_user_profile(self, user_id: int) -> Dict[str, Any]:
329 """Get user profile with error handling"""
330 try:
331 user = self.db.get_user(user_id)
332
333 if not user:
334 return {
335 "success": False,
336 "error_type": "not_found",
337 "message": f"User with ID {user_id} not found"
338 }
339
340 if not user.active:
341 return {
342 "success": False,
343 "error_type": "inactive_user",
344 "message": "User account is inactive"
345 }
346
347 return {
348 "success": True,
349 "user": {
350 "id": user.id,
351 "username": user.username,
352 "email": user.email,
353 "age": user.age,
354 "created_at": user.created_at.isoformat(),
355 "active": user.active
356 }
357 }
358
359 except Exception as e:
360 logging.error(f"Error retrieving user profile {user_id}: {e}")
361 return {
362 "success": False,
363 "error_type": "system_error",
364 "message": "Failed to retrieve user profile"
365 }
366
367def error_handler(func: Callable) -> Callable:
368 """Decorator for comprehensive error handling with logging"""
369
370 @functools.wraps(func)
371 def wrapper(*args, **kwargs) -> Any:
372 start_time = time.time()
373 function_name = func.__name__
374
375 try:
376 logging.debug(f"Starting {function_name} with args={args[:2]} kwargs={list(kwargs.keys())}")
377
378 result = func(*args, **kwargs)
379
380 execution_time = time.time() - start_time
381 logging.debug(f"Completed {function_name} in {execution_time:.3f}s")
382
383 return result
384
385 except Exception as e:
386 execution_time = time.time() - start_time
387
388 # Log error with context
389 error_context = {
390 "function": function_name,
391 "execution_time": execution_time,
392 "args_preview": str(args)[:100],
393 "error_type": type(e).__name__,
394 "error_message": str(e)
395 }
396
397 logging.error(f"Error in {function_name}: {e}")
398 logging.error(f"Error context: {error_context}")
399
400 # Re-raise the exception
401 raise
402
403 return wrapper
404
405@error_handler
406def demonstrate_user_operations():
407 """Demonstrate user operations with error handling"""
408 print("=== User Management System Demonstration ===\n")
409
410 # Setup
411 db_manager = DatabaseManager("demo_app.db")
412 user_service = UserService(db_manager)
413
414 # Test cases
415 test_users = [
416 ("alice_johnson", "[email protected]", 25), # Valid user
417 ("bob", "[email protected]", 30), # Valid user
418 ("", "[email protected]", 25), # Invalid username
419 ("carol", "invalid-email", 28), # Invalid email
420 ("david", "[email protected]", 22), # Blacklisted email
421 ("eve_smith", "[email protected]", 15), # Valid user
422 ("alice_johnson", "[email protected]", 26), # Duplicate username
423 ]
424
425 print("1. Testing user registration...")
426 created_users = []
427
428 for username, email, age in test_users:
429 print(f"\nRegistering: {username}, {email}, {age}")
430
431 try:
432 result = user_service.register_user(username, email, age)
433
434 if result["success"]:
435 print(f" ✅ Success: {result['message']}")
436 created_users.append(result["user_id"])
437 else:
438 print(f" ❌ Failed: {result['message']} ({result['error_type']})")
439
440 except Exception as e:
441 print(f" 💥 Unexpected error: {e}")
442
443 print(f"\nCreated {len(created_users)} users successfully")
444
445 # Test user retrieval
446 print("\n2. Testing user profile retrieval...")
447
448 test_user_ids = created_users + [999, 1000] # Include non-existent IDs
449
450 for user_id in test_user_ids:
451 print(f"\nRetrieving user ID {user_id}:")
452
453 try:
454 result = user_service.get_user_profile(user_id)
455
456 if result["success"]:
457 user = result["user"]
458 print(f" ✅ Found: {user['username']} ({user['email']})")
459 else:
460 print(f" ❌ Not found: {result['message']}")
461
462 except Exception as e:
463 print(f" 💥 Error: {e}")
464
465 # Test user updates
466 print("\n3. Testing user updates...")
467
468 if created_users:
469 user_id = created_users[0]
470
471 update_tests = [
472 {"username": "alice_updated"}, # Valid update
473 {"email": "[email protected]"}, # Valid update
474 {"age": 26}, # Valid update
475 {"age": 200}, # Invalid age
476 {"invalid_field": "value"}, # Invalid field
477 {"username": ""}, # Invalid username
478 ]
479
480 for updates in update_tests:
481 print(f"\nUpdating user {user_id} with {updates}:")
482
483 try:
484 result = db_manager.update_user(user_id, **updates)
485 print(f" ✅ Updated successfully")
486
487 except ValidationError as e:
488 print(f" ❌ Validation error: {e}")
489 except BusinessLogicError as e:
490 print(f" ❌ Business logic error: {e}")
491 except Exception as e:
492 print(f" 💥 Unexpected error: {e}")
493
494 print("\n4. Error statistics...")
495
496 # Demonstrate error logging analysis
497 log_file = "demo_app.log"
498 if os.path.exists(log_file):
499 with open(log_file, 'r') as f:
500 log_content = f.read()
501
502 error_count = log_content.count("ERROR")
503 warning_count = log_content.count("WARNING")
504 info_count = log_content.count("INFO")
505
506 print(f" Log entries: {info_count} INFO, {warning_count} WARNING, {error_count} ERROR")
507
508 print("\n=== Demonstration Complete ===")
509
510def setup_logging():
511 """Setup comprehensive logging configuration"""
512
513 # Create logs directory
514 logs_dir = Path("logs")
515 logs_dir.mkdir(exist_ok=True)
516
517 # Setup logging configuration
518 logging.basicConfig(
519 level=logging.DEBUG,
520 format='%(asctime)s | %(name)s | %(levelname)s | %(filename)s:%(lineno)d | %(message)s',
521 handlers=[
522 logging.FileHandler(logs_dir / "demo_app.log"),
523 logging.StreamHandler()
524 ]
525 )
526
527 # Set specific logger levels
528 logging.getLogger("sqlite3").setLevel(logging.WARNING)
529
530def main():
531 """Main demonstration function"""
532 try:
533 print("Robust Web Application Error Handling Demonstration")
534 print("=" * 60)
535
536 # Setup logging
537 setup_logging()
538
539 # Run demonstration
540 demonstrate_user_operations()
541
542 except KeyboardInterrupt:
543 print("\n\nDemonstration interrupted by user")
544
545 except Exception as e:
546 print(f"\n\nFatal error in main: {e}")
547 logging.critical(f"Fatal error in main: {e}")
548 logging.critical(f"Traceback: {traceback.format_exc()}")
549
550 finally:
551 print("\nDemonstration cleanup completed")
552
553if __name__ == "__main__":
554 main()Your progress is saved in this browser, on this device only.