Exception Handling Deep Dive in Python

The Basic try/except Block

# File: basics/try_except_basics.py

def divide_numbers(a, b):
    try:
        # Code that might raise an exception
        result = a / b
        return result
    except ZeroDivisionError:
        # Code that executes if a ZeroDivisionError occurs
        print("Error: Cannot divide by zero!")
        return None

# Test the function
print(divide_numbers(10, 2))  # Works fine: 5.0
print(divide_numbers(10, 0))  # Handles the error gracefully
                

Think of a try block as a safety net. You're telling Python: "Try to execute this code, and if anything goes wrong, I have a plan to handle it."

Handling Multiple Exception Types

# File: basics/multiple_except.py

def process_data(data):
    try:
        # Multiple things could go wrong here
        value = data['key']
        result = 100 / value
        return result
    except KeyError:
        # Handles missing dictionary key
        print("Error: The required key doesn't exist in the data!")
        return None
    except TypeError:
        # Handles case where data is not a dictionary
        print("Error: The data is not a dictionary!")
        return None
    except ZeroDivisionError:
        # Handles division by zero
        print("Error: Cannot divide by zero!")
        return None
    except Exception as e:
        # Catches any other exceptions
        print(f"Unexpected error: {e}")
        return None

# Test with different scenarios
print(process_data({'key': 4}))         # Works fine: 25.0
print(process_data({'different_key': 4}))  # KeyError
print(process_data("not a dict"))       # TypeError
print(process_data({'key': 0}))         # ZeroDivisionError
print(process_data({'key': '4'}))       # TypeError (can't divide by string)
                

You can catch multiple exception types by providing multiple except blocks. Python will check each exception type in order and execute the first matching handler.

Using else and finally

# File: basics/else_finally.py

def read_file(filename):
    try:
        file = open(filename, 'r')
        content = file.read()
    except FileNotFoundError:
        print(f"Error: The file '{filename}' was not found.")
        return None
    except PermissionError:
        print(f"Error: You don't have permission to read '{filename}'.")
        return None
    except Exception as e:
        print(f"Unexpected error while reading the file: {e}")
        return None
    else:
        # This code executes if no exceptions were raised in the try block
        print(f"Successfully read {len(content)} characters from '{filename}'.")
        return content
    finally:
        # This code always executes, regardless of whether an exception occurred
        print("Cleanup: Ensuring file is closed.")
        try:
            file.close()
        except:
            # Handle case where file was never successfully opened
            pass

# Test with different scenarios
content = read_file("existing_file.txt")  # Successful case
content = read_file("nonexistent_file.txt")  # FileNotFoundError
                

The else block runs only if no exceptions were raised in the try block. This is useful for code that should only execute if the try block was successful.

The finally block always runs, regardless of whether an exception occurred. This is perfect for cleanup operations like closing files, network connections, or releasing other resources.

Basic Exception Propagation

# File: basics/exception_propagation.py

def level3():
    # This function raises an exception
    x = 1 / 0  # ZeroDivisionError
    return x

def level2():
    # This function calls level3
    print("Entering level2")
    result = level3()
    print("Exiting level2")  # This line never executes if level3 raises an exception
    return result

def level1():
    # This function calls level2, but catches exceptions
    print("Entering level1")
    try:
        result = level2()
        return result
    except ZeroDivisionError:
        print("Caught divide by zero error in level1")
        return None
    finally:
        print("Exiting level1")

# Main program
print("Starting program")
result = level1()
print(f"Final result: {result}")
print("Program completed")

# Output:
# Starting program
# Entering level1
# Entering level2
# Caught divide by zero error in level1
# Exiting level1
# Final result: None
# Program completed
                

In this example, the ZeroDivisionError originates in level3(), but it's not handled there. It propagates up to level2(), which also doesn't handle it. Finally, it reaches level1(), which has a try/except block that catches it.

This is like a hot potato being passed up a line of people until someone has gloves to handle it.

Partial Handling and Re-raising

# File: basics/exception_reraising.py

def process_data(data):
    try:
        value = 100 / data
        return value
    except ZeroDivisionError:
        # We'll handle this partially, then re-raise
        print("Warning: Attempted to divide by zero!")
        # We can do some cleanup or logging here
        
        # Then re-raise the exception to let the caller handle it further
        raise  # Re-raises the current exception

def main():
    try:
        result = process_data(0)
        print(f"Result: {result}")
    except ZeroDivisionError:
        print("Error handled in main: Cannot divide by zero")
        # Provide a default value or alternative
        result = float('inf')  # Use infinity as a fallback
        print(f"Using fallback value: {result}")

# Run the program
main()

# Output:
# Warning: Attempted to divide by zero!
# Error handled in main: Cannot divide by zero
# Using fallback value: inf
                

Sometimes, you may want to handle an exception partially at one level, then let a higher level handle it further. You can do this by re-raising the exception using the raise statement without arguments.

Catching Exceptions at Different Levels

# File: basics/multilevel_exception_handling.py

def parse_data(data_string):
    try:
        # Attempt to parse as an integer
        return int(data_string)
    except ValueError:
        # If that fails, try to parse as float
        try:
            return float(data_string)
        except ValueError:
            # If that also fails, raise a custom error
            raise ValueError(f"Cannot parse '{data_string}' as a number")

def process_input(input_value):
    try:
        number = parse_data(input_value)
        result = 100 / number
        return f"Processing complete: {result}"
    except ValueError as e:
        # Handle parsing errors
        return f"Input error: {e}"
    except ZeroDivisionError:
        # Handle division by zero
        return "Error: Cannot divide by zero"

# Test with different inputs
inputs = ["42", "3.14", "zero", "0"]
for input_value in inputs:
    output = process_input(input_value)
    print(f"Input: {input_value} -> {output}")

# Output:
# Input: 42 -> Processing complete: 2.380952380952381
# Input: 3.14 -> Processing complete: 31.847133757961785
# Input: zero -> Input error: Cannot parse 'zero' as a number
# Input: 0 -> Error: Cannot divide by zero
                

This example shows how exceptions can be caught and handled at different levels in your program, based on where it makes the most sense to handle each type of error.

Basic Custom Exceptions

# File: custom_exceptions/custom_exception_classes.py

# Define custom exception classes
class InsufficientFundsError(Exception):
    """Raised when a bank account doesn't have enough money for a withdrawal."""
    pass

class InvalidAmountError(Exception):
    """Raised when an invalid monetary amount is provided."""
    pass

class AccountLockedError(Exception):
    """Raised when an operation is attempted on a locked account."""
    pass

# A simple bank account class using custom exceptions
class BankAccount:
    def __init__(self, account_id, balance=0):
        self.account_id = account_id
        self.balance = balance
        self.is_locked = False
    
    def deposit(self, amount):
        if self.is_locked:
            raise AccountLockedError(f"Account {self.account_id} is locked")
        
        if amount <= 0:
            raise InvalidAmountError(f"Deposit amount must be positive, got {amount}")
        
        self.balance += amount
        return self.balance
    
    def withdraw(self, amount):
        if self.is_locked:
            raise AccountLockedError(f"Account {self.account_id} is locked")
        
        if amount <= 0:
            raise InvalidAmountError(f"Withdrawal amount must be positive, got {amount}")
        
        if amount > self.balance:
            raise InsufficientFundsError(
                f"Cannot withdraw ${amount}. Account balance is ${self.balance}"
            )
        
        self.balance -= amount
        return self.balance
    
    def lock_account(self):
        self.is_locked = True
    
    def unlock_account(self):
        self.is_locked = False

# Usage example
def demonstrate_custom_exceptions():
    account = BankAccount("12345", 100)
    
    try:
        account.deposit(50)
        print(f"Balance after deposit: ${account.balance}")
        
        account.withdraw(30)
        print(f"Balance after withdrawal: ${account.balance}")
        
        # This should raise an InsufficientFundsError
        account.withdraw(200)
    except InsufficientFundsError as e:
        print(f"Error: {e}")
    
    try:
        # This should raise an InvalidAmountError
        account.deposit(-50)
    except InvalidAmountError as e:
        print(f"Error: {e}")
    
    account.lock_account()
    try:
        # This should raise an AccountLockedError
        account.withdraw(10)
    except AccountLockedError as e:
        print(f"Error: {e}")

# Run the demonstration
demonstrate_custom_exceptions()
                

Creating custom exceptions makes your code more expressive and helps users of your code understand what went wrong. It's like creating a specialized language for the errors in your domain.

Custom Exception Hierarchy

# File: custom_exceptions/exception_hierarchy.py

# Base exception for our application
class AppError(Exception):
    """Base exception for the application."""
    pass

# Database error hierarchy
class DatabaseError(AppError):
    """Base exception for database-related errors."""
    pass

class ConnectionError(DatabaseError):
    """Raised when database connection fails."""
    pass

class QueryError(DatabaseError):
    """Raised when a database query fails."""
    pass

# Validation error hierarchy
class ValidationError(AppError):
    """Base exception for validation-related errors."""
    pass

class RequiredFieldError(ValidationError):
    """Raised when a required field is missing."""
    def __init__(self, field_name):
        self.field_name = field_name
        super().__init__(f"Required field missing: {field_name}")

class InvalidFormatError(ValidationError):
    """Raised when a field has an invalid format."""
    def __init__(self, field_name, expected_format):
        self.field_name = field_name
        self.expected_format = expected_format
        super().__init__(
            f"Field '{field_name}' has invalid format. Expected: {expected_format}"
        )

# Authentication error hierarchy
class AuthError(AppError):
    """Base exception for authentication-related errors."""
    pass

class LoginFailedError(AuthError):
    """Raised when login fails."""
    pass

class PermissionDeniedError(AuthError):
    """Raised when user doesn't have permission for an action."""
    pass

# Example usage with a database connection function
def connect_to_database(connection_string):
    try:
        # Simulate connection logic
        if not connection_string:
            raise ConnectionError("Empty connection string")
        
        if "invalid" in connection_string:
            raise ConnectionError(f"Invalid connection string: {connection_string}")
        
        print(f"Connected to database with {connection_string}")
        return {"connection": "object"}
    
    except ConnectionError:
        # Re-raise the exception
        raise
    except Exception as e:
        # Wrap any other exceptions in our custom type
        raise DatabaseError(f"Unexpected database error: {e}")

# Example validation function
def validate_user(user_data):
    required_fields = ['username', 'email', 'password']
    
    # Check required fields
    for field in required_fields:
        if field not in user_data or not user_data[field]:
            raise RequiredFieldError(field)
    
    # Validate email format (simplified)
    if '@' not in user_data['email']:
        raise InvalidFormatError('email', 'valid email address')
    
    # Validate password length
    if len(user_data['password']) < 8:
        raise InvalidFormatError('password', 'at least 8 characters')
    
    return True

# Example usage
def process_user_registration(user_data):
    try:
        # Validate user data
        validate_user(user_data)
        
        # Connect to database
        db = connect_to_database("mysql://validconnection")
        
        # Additional processing would go here
        print(f"User {user_data['username']} registered successfully")
        
    except ValidationError as e:
        print(f"Validation failed: {e}")
        # Log error, display to user, etc.
        
    except DatabaseError as e:
        print(f"Database error: {e}")
        # Log error, retry, or show system error
        
    except AppError as e:
        print(f"Application error: {e}")
        # Generic application error handling

# Test with valid and invalid data
valid_user = {
    'username': 'johndoe',
    'email': 'john@example.com',
    'password': 'securepass123'
}

invalid_user = {
    'username': 'janedoe',
    'email': 'invalid-email',
    'password': 'short'
}

missing_field_user = {
    'username': 'bobsmith',
    # missing email
    'password': 'securepw456'
}

process_user_registration(valid_user)
process_user_registration(invalid_user)
process_user_registration(missing_field_user)
                

Creating a hierarchy of custom exceptions allows you to catch exceptions at different levels of specificity. It's like having categories and subcategories for different types of errors.

Adding Context to Custom Exceptions

# File: custom_exceptions/enhanced_exceptions.py

class DataProcessingError(Exception):
    """Base exception for data processing errors."""
    
    def __init__(self, message, data=None, source=None):
        self.data = data
        self.source = source
        self.message = message
        
        # Create a more informative error message with context
        error_msg = message
        if source:
            error_msg += f" (Source: {source})"
        
        super().__init__(error_msg)
    
    def get_context(self):
        """Return a dictionary with all context information."""
        return {
            'message': self.message,
            'data': self.data,
            'source': self.source
        }

class DataParsingError(DataProcessingError):
    """Raised when data cannot be parsed correctly."""
    
    def __init__(self, message, data=None, source=None, line_number=None):
        self.line_number = line_number
        
        # Include line number in the error message if available
        if line_number is not None:
            message = f"{message} at line {line_number}"
        
        super().__init__(message, data, source)
    
    def get_context(self):
        context = super().get_context()
        context['line_number'] = self.line_number
        return context

# Example usage
def parse_data_file(filename):
    try:
        with open(filename, 'r') as file:
            for i, line in enumerate(file, 1):
                try:
                    # Simulate parsing logic
                    parts = line.strip().split(',')
                    
                    if len(parts) < 2:
                        raise DataParsingError(
                            "Insufficient data fields",
                            data=line.strip(),
                            source=filename,
                            line_number=i
                        )
                    
                    # Process the data
                    print(f"Processed line {i}: {parts}")
                    
                except DataParsingError:
                    # Re-raise parsing errors
                    raise
                except Exception as e:
                    # Wrap other errors in our custom exception
                    raise DataParsingError(
                        f"Error parsing data: {e}",
                        data=line.strip(),
                        source=filename,
                        line_number=i
                    )
    
    except FileNotFoundError:
        raise DataProcessingError(f"File not found", source=filename)

# Error handling function
def process_with_detailed_errors():
    try:
        parse_data_file("sample_data.csv")
    except DataParsingError as e:
        print(f"Parsing error: {e}")
        context = e.get_context()
        print("Error context:")
        for key, value in context.items():
            print(f"  {key}: {value}")
    except DataProcessingError as e:
        print(f"Processing error: {e}")
        context = e.get_context()
        print("Error context:")
        for key, value in context.items():
            print(f"  {key}: {value}")

process_with_detailed_errors()
                

Enhancing custom exceptions with additional context information makes debugging easier and provides more helpful error messages to users of your code.

Exception Chaining with raise from

# File: advanced/exception_chaining.py

def fetch_data_from_database(query):
    try:
        # Simulate a database operation
        if "SELECT" not in query.upper():
            raise ValueError("Invalid query: must be a SELECT statement")
        
        # Simulate connectivity issues
        if "users" in query.lower():
            raise ConnectionError("Database connection timeout")
        
        return ["result1", "result2"]
    
    except (ValueError, ConnectionError) as e:
        # Chain the original exception to a new, more specific one
        raise DatabaseQueryError(f"Failed to execute query: {query}") from e

class DatabaseQueryError(Exception):
    """Raised when a database query fails."""
    pass

def get_user_data(user_id):
    try:
        query = f"SELECT * FROM users WHERE id = {user_id}"
        return fetch_data_from_database(query)
    except DatabaseQueryError as e:
        print(f"Error: {e}")
        # Access the original exception that caused this one
        original_error = e.__cause__
        print(f"Caused by: {original_error} (type: {type(original_error).__name__})")
        return None

# Test the function
result = get_user_data(42)
print(f"Result: {result}")

# Example with exception not using 'from'
def bad_exception_chaining():
    try:
        # Try something that might fail
        x = 1 / 0
    except ZeroDivisionError:
        # Raising a new exception without chaining loses the original cause
        raise RuntimeError("Something went wrong!")

try:
    bad_exception_chaining()
except RuntimeError as e:
    print(f"Error: {e}")
    # This will be None because we didn't use 'raise ... from'
    print(f"Caused by: {e.__cause__}")
                

The raise from syntax explicitly chains exceptions, preserving the original cause of an error. This is invaluable for debugging complex applications.

Think of it like leaving a trail of breadcrumbs—instead of replacing one error with another, you're creating a chain that leads back to the original problem.

Using Context Managers for Cleanup

# File: advanced/cleanup_actions.py

class Resource:
    def __init__(self, name):
        self.name = name
        print(f"Resource {name} initialized")
        
    def read(self):
        print(f"Reading from resource {self.name}")
        return f"Data from {self.name}"
        
    def close(self):
        print(f"Resource {self.name} closed")

# Traditional approach with try/finally
def use_resource_traditional(name):
    resource = Resource(name)
    try:
        data = resource.read()
        return data
    finally:
        resource.close()

# Context manager approach
class ResourceManager:
    def __init__(self, name):
        self.name = name
        self.resource = None
        
    def __enter__(self):
        self.resource = Resource(self.name)
        return self.resource
        
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.resource:
            self.resource.close()
        
        # Return True to suppress the exception, False to propagate it
        return False  # Let any exceptions propagate

def use_resource_context_manager(name):
    with ResourceManager(name) as resource:
        data = resource.read()
        return data

# Using a context manager for error handling and cleanup
def process_resource(name, use_context_manager=True):
    try:
        if use_context_manager:
            print("\nUsing context manager:")
            data = use_resource_context_manager(name)
        else:
            print("\nUsing traditional approach:")
            data = use_resource_traditional(name)
            
        print(f"Processed data: {data}")
    except Exception as e:
        print(f"Error processing resource: {e}")

# Test both approaches
process_resource("file1", use_context_manager=False)
process_resource("file2", use_context_manager=True)

# Example with built-in context manager
def process_file(filename):
    try:
        with open(filename, 'r') as file:
            content = file.read()
            return content
    except FileNotFoundError:
        print(f"File not found: {filename}")
        return None

content = process_file("existing_file.txt")
print(f"File content: {content}")
                

Context managers (using the with statement) provide a clean way to ensure that resources are properly managed, even if exceptions occur. They're perfect for files, network connections, database transactions, and other resources that need cleanup.

Traceback Manipulation

# File: advanced/traceback_manipulation.py

import traceback
import sys

def print_custom_traceback(e):
    """Print a custom traceback with selective information."""
    print("\n--- Custom Traceback ---")
    tb_list = traceback.extract_tb(e.__traceback__)
    
    for i, (filename, line, func, text) in enumerate(tb_list):
        # Skip certain internal frames if desired
        if "lib/python" in filename and "internal_module" in filename:
            continue
            
        print(f"Frame {i}: File '{filename}', line {line}, in {func}")
        print(f"  Code: {text}")
    
    print(f"Exception: {type(e).__name__}: {e}")
    print("-----------------------\n")

def capture_exception_info():
    """Capture and return exception information without raising."""
    exc_type, exc_value, exc_traceback = sys.exc_info()
    return {
        "type": exc_type.__name__ if exc_type else None,
        "message": str(exc_value) if exc_value else None,
        "traceback": traceback.format_tb(exc_traceback) if exc_traceback else None
    }

def function_c():
    # This will raise a ZeroDivisionError
    return 1 / 0

def function_b():
    return function_c()

def function_a():
    return function_b()

def demonstrate_traceback_handling():
    try:
        function_a()
    except Exception as e:
        # Standard traceback printing
        print("Standard traceback:")
        traceback.print_exc()
        
        # Custom traceback printing
        print_custom_traceback(e)
        
        # Capturing exception info as a dictionary
        error_info = capture_exception_info()
        print("Captured error info as dictionary:")
        for key, value in error_info.items():
            print(f"{key}: {value}")

demonstrate_traceback_handling()
                

Python's traceback module allows you to extract, format, and manipulate the traceback information from exceptions. This is useful for creating custom error reports or logging systems.

The Retry Pattern

# File: patterns/retry_pattern.py

import time
import random

class TooManyRetriesError(Exception):
    """Raised when an operation fails after multiple retry attempts."""
    pass

def retry(func, max_attempts=3, delay=1, backoff_factor=2, exceptions_to_catch=(Exception,)):
    """
    Retry a function call with exponential backoff.
    
    Args:
        func: The function to call
        max_attempts: Maximum number of retry attempts
        delay: Initial delay between retries (in seconds)
        backoff_factor: Factor by which the delay increases each retry
        exceptions_to_catch: Tuple of exception types to catch and retry
    
    Returns:
        The result of the function call if successful
    
    Raises:
        TooManyRetriesError: If the function fails after max_attempts
    """
    attempt = 1
    last_exception = None
    
    while attempt <= max_attempts:
        try:
            print(f"Attempt {attempt}/{max_attempts}...")
            return func()
        except exceptions_to_catch as e:
            last_exception = e
            
            if attempt == max_attempts:
                # We've reached the maximum attempts, so give up
                break
                
            # Calculate wait time with exponential backoff
            wait_time = delay * (backoff_factor ** (attempt - 1))
            
            # Add some randomness to avoid thundering herd problem
            jitter = random.uniform(0, 0.1 * wait_time)
            wait_time += jitter
            
            print(f"Attempt {attempt} failed: {e}")
            print(f"Retrying in {wait_time:.2f} seconds...")
            
            time.sleep(wait_time)
            attempt += 1
    
    # If we got here, all attempts failed
    raise TooManyRetriesError(f"Failed after {max_attempts} attempts. Last error: {last_exception}")

# Example usage with an unreliable network function
def unreliable_network_call():
    """Simulate an unreliable network call that sometimes fails."""
    # Simulate random failures
    if random.random() < 0.7:  # 70% chance of failure
        raise ConnectionError("Network timeout")
    
    return "Data from network"

# Example with a database connection function
def unreliable_database_query():
    """Simulate an unreliable database query."""
    # Simulate different types of failures
    failure_type = random.randint(1, 3)
    
    if failure_type == 1:
        raise ConnectionError("Database connection failed")
    elif failure_type == 2:
        raise TimeoutError("Query timeout")
    elif failure_type == 3 and random.random() < 0.5:
        # 50% chance of this type of failure actually happening
        raise PermissionError("Insufficient permissions")
    
    return ["result1", "result2", "result3"]

# Try the retry pattern
try:
    # Retry the unreliable network call
    result1 = retry(unreliable_network_call, max_attempts=5, delay=0.5)
    print(f"Network result: {result1}")
    
    # Retry the database query, only catching specific exceptions
    result2 = retry(
        unreliable_database_query,
        max_attempts=4,
        exceptions_to_catch=(ConnectionError, TimeoutError)
    )
    print(f"Database result: {result2}")
    
except TooManyRetriesError as e:
    print(f"Operation failed: {e}")
                

The retry pattern is essential for handling transient errors in distributed systems, like network hiccups or temporary service unavailability.

The Circuit Breaker Pattern

# File: patterns/circuit_breaker.py

import time
from enum import Enum, auto

class CircuitState(Enum):
    CLOSED = auto()  # Normal operation, requests pass through
    OPEN = auto()    # Not allowing requests through
    HALF_OPEN = auto()  # Testing if the service is back to normal

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=10, expected_exceptions=(Exception,)):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.expected_exceptions = expected_exceptions
        
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        
    def execute(self, func, *args, **kwargs):
        """Execute the function with circuit breaker protection."""
        
        if self.state == CircuitState.OPEN:
            # Check if recovery timeout has elapsed
            if time.time() - self.last_failure_time >= self.recovery_timeout:
                print("Circuit half-open, testing service...")
                self.state = CircuitState.HALF_OPEN
            else:
                raise CircuitBreakerOpenError(
                    f"Circuit is open. Service unavailable for another "
                    f"{self.recovery_timeout - (time.time() - self.last_failure_time):.1f} seconds"
                )
        
        try:
            result = func(*args, **kwargs)
            
            # If the request succeeded and we were in half-open state,
            # reset the circuit breaker
            if self.state == CircuitState.HALF_OPEN:
                print("Service recovered, closing circuit")
                self.reset()
                
            return result
            
        except self.expected_exceptions as e:
            self._handle_failure(e)
            raise
    
    def _handle_failure(self, exception):
        """Handle a service failure."""
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.state == CircuitState.CLOSED and self.failure_count >= self.failure_threshold:
            print(f"Failure threshold reached ({self.failure_count}), opening circuit")
            self.state = CircuitState.OPEN
            
        elif self.state == CircuitState.HALF_OPEN:
            print("Service still failing in half-open state, reopening circuit")
            self.state = CircuitState.OPEN
    
    def reset(self):
        """Reset the circuit breaker to its initial state."""
        self.failure_count = 0
        self.state = CircuitState.CLOSED
        
    def __str__(self):
        return f"CircuitBreaker(state={self.state.name}, failures={self.failure_count})"

class CircuitBreakerOpenError(Exception):
    """Raised when a service call is attempted while the circuit is open."""
    pass

# Example with an unreliable service
def unreliable_service():
    """Simulate an unreliable service that fails sometimes."""
    import random
    
    # 80% chance of failure during the first 10 calls, then recovers
    global service_call_count
    service_call_count += 1
    
    failure_probability = 0.8 if service_call_count < 10 else 0.1
    
    if random.random() < failure_probability:
        raise ConnectionError("Service unavailable")
    
    return "Service response data"

# Initialize the global counter
service_call_count = 0

# Create a circuit breaker
breaker = CircuitBreaker(
    failure_threshold=3,
    recovery_timeout=5,
    expected_exceptions=(ConnectionError, TimeoutError)
)

# Try making several calls to the service
for i in range(15):
    print(f"\nCall {i+1}:")
    try:
        result = breaker.execute(unreliable_service)
        print(f"Service call successful: {result}")
    except CircuitBreakerOpenError as e:
        print(f"Circuit breaker prevented call: {e}")
    except Exception as e:
        print(f"Service call failed: {e}")
    
    print(f"Circuit breaker status: {breaker}")
    
    # Short delay between calls
    time.sleep(1)
                

The circuit breaker pattern prevents cascading failures by stopping calls to a failing service after too many errors, then periodically testing if the service has recovered.

It's like a circuit breaker in your home's electrical system—when there's a problem, it trips to prevent further damage, then later it can be reset.

The Transaction Pattern

# File: patterns/transaction_pattern.py

class TransactionError(Exception):
    """Base exception for transaction-related errors."""
    pass

class Transaction:
    """Simple transaction context manager."""
    
    def __init__(self, *resources):
        self.resources = resources
        self.completed = False
    
    def __enter__(self):
        # Begin the transaction
        print("Beginning transaction")
        try:
            for resource in self.resources:
                if hasattr(resource, 'begin'):
                    resource.begin()
            return self
        except Exception as e:
            # Rollback if beginning fails
            self.rollback()
            raise TransactionError(f"Failed to begin transaction: {e}") from e
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # If an exception occurred, rollback
        if exc_type is not None:
            print(f"Transaction failed: {exc_val}")
            self.rollback()
            return False  # Let the exception propagate
        
        # If we got here, commit the transaction
        if not self.completed:
            self.commit()
        
        return False  # Let any exceptions propagate
    
    def commit(self):
        """Commit all resources in the transaction."""
        try:
            print("Committing transaction")
            for resource in self.resources:
                if hasattr(resource, 'commit'):
                    resource.commit()
            self.completed = True
        except Exception as e:
            # If commit fails, try to rollback
            self.rollback()
            raise TransactionError(f"Failed to commit transaction: {e}") from e
    
    def rollback(self):
        """Rollback all resources in the transaction."""
        print("Rolling back transaction")
        rollback_exceptions = []
        
        # Try to rollback all resources, collecting any exceptions
        for resource in self.resources:
            if hasattr(resource, 'rollback'):
                try:
                    resource.rollback()
                except Exception as e:
                    rollback_exceptions.append(e)
        
        # If any rollbacks failed, raise a new exception
        if rollback_exceptions:
            raise TransactionError(
                f"Failed to rollback transaction: {rollback_exceptions}"
            )

# Example resource classes
class Database:
    def __init__(self, name):
        self.name = name
        print(f"Database {name} initialized")
    
    def begin(self):
        print(f"Database {self.name}: Beginning transaction")
    
    def commit(self):
        print(f"Database {self.name}: Committing changes")
    
    def rollback(self):
        print(f"Database {self.name}: Rolling back changes")
    
    def execute(self, query):
        print(f"Database {self.name}: Executing query: {query}")

class FileSystem:
    def __init__(self, root_dir):
        self.root_dir = root_dir
        self.temp_files = []
        print(f"FileSystem at {root_dir} initialized")
    
    def begin(self):
        print(f"FileSystem: Beginning transaction")
    
    def commit(self):
        print(f"FileSystem: Committing {len(self.temp_files)} temporary files")
        self.temp_files = []
    
    def rollback(self):
        print(f"FileSystem: Deleting {len(self.temp_files)} temporary files")
        self.temp_files = []
    
    def write_file(self, filename, content):
        print(f"FileSystem: Writing to {filename}")
        self.temp_files.append(filename)

# Example usage
def perform_data_migration(success=True):
    # Create resources
    db = Database("UserDB")
    fs = FileSystem("/var/data")
    
    # Use transaction pattern
    with Transaction(db, fs) as transaction:
        # Perform operations
        db.execute("SELECT * FROM users")
        db.execute("UPDATE users SET status = 'active'")
        
        fs.write_file("user_export.csv", "user data")
        fs.write_file("log.txt", "migration completed")
        
        # Simulate an error if success=False
        if not success:
            raise ValueError("Simulated error during migration")
        
        # If we get here and success=True, the transaction will be committed

# Test successful transaction
print("\nSuccessful transaction:")
perform_data_migration(success=True)

# Test failed transaction
print("\nFailed transaction:")
try:
    perform_data_migration(success=False)
except Exception as e:
    print(f"Caught exception: {e}")
                

The transaction pattern ensures that a series of operations either all succeed or all fail, with no partial changes. This is crucial for maintaining data consistency.

Logging Exceptions

# File: advanced/exception_logging.py

import logging
import traceback
import sys
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("app.log"),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger("ExceptionExample")

def log_exception(e, log_level=logging.ERROR, include_traceback=True):
    """Log an exception with optional traceback."""
    exception_type = type(e).__name__
    
    # Basic exception info
    msg = f"Exception {exception_type}: {e}"
    
    # Add traceback if requested
    if include_traceback:
        exception_traceback = traceback.format_exc()
        msg += f"\n{exception_traceback}"
    
    # Log at the specified level
    logger.log(log_level, msg)

def log_cause_chain(e, log_level=logging.ERROR):
    """Log an exception and its causes."""
    current = e
    chain = []
    
    # Build the cause chain
    while current:
        chain.append((type(current).__name__, str(current)))
        current = current.__cause__
    
    # Create message
    msg_parts = ["Exception chain:"]
    for i, (exc_type, exc_msg) in enumerate(chain):
        prefix = "└─ " if i == 0 else f"   {'  ' * i}└─ "
        msg_parts.append(f"{prefix}{exc_type}: {exc_msg}")
    
    msg = "\n".join(msg_parts)
    logger.log(log_level, msg)

def function_that_logs_exceptions():
    try:
        result = 1 / 0
    except Exception as e:
        # Basic logging
        logger.error(f"An error occurred: {e}")
        
        # More detailed logging
        log_exception(e)
        
        # Re-raise if needed
        raise

def demonstrate_logging():
    # Example 1: Simple exception logging
    try:
        logger.info("Running example 1...")
        function_that_logs_exceptions()
    except Exception as e:
        logger.info("Example 1 completed with expected exception")
    
    # Example 2: Exception chaining and cause logging
    try:
        logger.info("Running example 2...")
        try:
            # This will raise a TypeError
            int("not_a_number")
        except TypeError as e:
            # Chain it to a higher-level exception
            raise ValueError("Failed to process input") from e
    except Exception as e:
        # Log the chain of causes
        log_cause_chain(e)
        logger.info("Example 2 completed with expected exception")
    
    # Example 3: Logging with context
    try:
        logger.info("Running example 3...")
        user_id = "user123"
        operation = "account_update"
        timestamp = datetime.now().isoformat()
        
        # Simulate an error
        raise PermissionError("Access denied")
    except Exception as e:
        # Log with context
        logger.error(
            f"Operation failed: {e}",
            extra={
                "user_id": user_id,
                "operation": operation,
                "timestamp": timestamp
            }
        )
        logger.info("Example 3 completed with expected exception")

# Run the demonstration
demonstrate_logging()
                

Proper exception logging is crucial for troubleshooting issues in production. It provides the context and information needed to understand what went wrong.

Debugging with Assertions

# File: advanced/debugging_with_assertions.py

def calculate_average(numbers):
    """Calculate the average of a list of numbers."""
    # Use assertions to validate inputs during development
    assert isinstance(numbers, list), f"Expected list, got {type(numbers)}"
    assert numbers, "List cannot be empty"
    assert all(isinstance(n, (int, float)) for n in numbers), "All elements must be numbers"
    
    return sum(numbers) / len(numbers)

def process_user_data(user_data):
    """Process user data with assertions for debugging."""
    # Check data structure
    assert isinstance(user_data, dict), "User data must be a dictionary"
    
    # Check required fields
    assert "name" in user_data, "Missing 'name' field"
    assert "age" in user_data, "Missing 'age' field"
    
    # Check data types
    assert isinstance(user_data["name"], str), "Name must be a string"
    assert isinstance(user_data["age"], int), "Age must be an integer"
    
    # Check value constraints
    assert 0 <= user_data["age"] <= 120, "Age must be between 0 and 120"
    
    # Process the data
    print(f"Processing user: {user_data['name']}, {user_data['age']} years old")
    
    return {
        "processed": True,
        "user_id": hash(user_data["name"]) % 10000,
        "category": "adult" if user_data["age"] >= 18 else "minor"
    }

def demonstrate_assertions():
    # Example 1: Valid input
    try:
        average = calculate_average([1, 2, 3, 4, 5])
        print(f"Average: {average}")
    except AssertionError as e:
        print(f"Assertion failed: {e}")
    
    # Example 2: Invalid input (empty list)
    try:
        average = calculate_average([])
        print(f"Average: {average}")
    except AssertionError as e:
        print(f"Assertion failed: {e}")
    
    # Example 3: Invalid input (wrong type)
    try:
        average = calculate_average("not a list")
        print(f"Average: {average}")
    except AssertionError as e:
        print(f"Assertion failed: {e}")
    
    # Example 4: Valid user data
    try:
        result = process_user_data({"name": "Alice", "age": 30})
        print(f"Result: {result}")
    except AssertionError as e:
        print(f"Assertion failed: {e}")
    
    # Example 5: Invalid user data
    try:
        result = process_user_data({"name": "Bob"})  # Missing age
        print(f"Result: {result}")
    except AssertionError as e:
        print(f"Assertion failed: {e}")

# Assertions can be disabled with the -O flag when running Python
# If they're performance-critical, use explicit checks instead

# In a production environment, you might have code like this:
def calculate_average_production(numbers):
    """Production-safe version of calculate_average."""
    if not isinstance(numbers, list):
        raise TypeError(f"Expected list, got {type(numbers)}")
    if not numbers:
        raise ValueError("List cannot be empty")
    if not all(isinstance(n, (int, float)) for n in numbers):
        raise TypeError("All elements must be numbers")
    
    return sum(numbers) / len(numbers)

# Run the demonstration
demonstrate_assertions()
                

Assertions are a great tool during development for catching logical errors and validating assumptions. Unlike regular exception handling, they're meant to catch programmer errors rather than runtime conditions.

Creating Data-Rich Exception Classes

        # File: custom_exceptions/data_rich_exceptions.py
        
        class ValidationError(Exception):
            """Base exception for data validation errors."""
            
            def __init__(self, message, field=None, value=None, constraints=None):
                self.field = field
                self.value = value
                self.constraints = constraints or {}
                
                # Enhance the message with field information if provided
                full_message = message
                if field:
                    full_message = f"{message} (field: {field}"
                    if value is not None:
                        full_message += f", value: {repr(value)}"
                    full_message += ")"
                
                super().__init__(full_message)
            
            def as_dict(self):
                """Return the error details as a dictionary."""
                return {
                    'error_type': self.__class__.__name__,
                    'message': str(self),
                    'field': self.field,
                    'value': self.value,
                    'constraints': self.constraints
                }
        
        class RequiredFieldError(ValidationError):
            """Raised when a required field is missing or empty."""
            
            def __init__(self, field, message=None):
                default_message = f"Required field missing or empty"
                super().__init__(message or default_message, field=field)
        
        class TypeMismatchError(ValidationError):
            """Raised when a field's value is not of the expected type."""
            
            def __init__(self, field, value, expected_type, message=None):
                default_message = f"Expected type {expected_type.__name__}"
                super().__init__(
                    message or default_message,
                    field=field,
                    value=value,
                    constraints={'expected_type': expected_type.__name__}
                )
                self.expected_type = expected_type
        
        class ValueRangeError(ValidationError):
            """Raised when a numeric value is outside the allowed range."""
            
            def __init__(self, field, value, min_value=None, max_value=None, message=None):
                constraints = {}
                if min_value is not None:
                    constraints['min_value'] = min_value
                if max_value is not None:
                    constraints['max_value'] = max_value
                
                range_str = []
                if min_value is not None:
                    range_str.append(f">= {min_value}")
                if max_value is not None:
                    range_str.append(f"<= {max_value}")
                
                default_message = f"Value outside allowed range ({' and '.join(range_str)})"
                
                super().__init__(
                    message or default_message,
                    field=field,
                    value=value,
                    constraints=constraints
                )
                self.min_value = min_value
                self.max_value = max_value
        
        class PatternMismatchError(ValidationError):
            """Raised when a string value doesn't match the required pattern."""
            
            def __init__(self, field, value, pattern, message=None):
                default_message = f"Value doesn't match required pattern"
                super().__init__(
                    message or default_message,
                    field=field,
                    value=value,
                    constraints={'pattern': str(pattern)}
                )
                self.pattern = pattern
        
        # Example usage with a data validation function
        def validate_user_data(data):
            """
            Validate a user data dictionary with rich error reporting.
            
            Args:
                data: Dictionary containing user data
            
            Returns:
                True if validation passes
            
            Raises:
                ValidationError: If any validation check fails
            """
            # Check for required fields
            required_fields = ['username', 'email', 'age']
            for field in required_fields:
                if field not in data or not data[field]:
                    raise RequiredFieldError(field)
            
            # Validate types
            if not isinstance(data['username'], str):
                raise TypeMismatchError('username', data['username'], str)
            
            if not isinstance(data['email'], str):
                raise TypeMismatchError('email', data['email'], str)
            
            if not isinstance(data['age'], int):
                raise TypeMismatchError('age', data['age'], int)
            
            # Validate value ranges
            if data['age'] < 18 or data['age'] > 120:
                raise ValueRangeError('age', data['age'], min_value=18, max_value=120)
            
            # Validate patterns (simplified email check)
            if '@' not in data['email']:
                raise PatternMismatchError(
                    'email', 
                    data['email'], 
                    'user@example.com',
                    "Invalid email format"
                )
            
            return True
        
        # Function to handle validation errors and create user-friendly messages
        def get_validation_errors(data):
            """
            Validate data and return user-friendly error messages.
            
            Args:
                data: Dictionary containing data to validate
            
            Returns:
                A list of error messages, or an empty list if validation passes
            """
            try:
                validate_user_data(data)
                return []
            except ValidationError as e:
                # Create user-friendly error message
                if isinstance(e, RequiredFieldError):
                    return [f"Please provide your {e.field}."]
                elif isinstance(e, TypeMismatchError):
                    return [f"The {e.field} field has an invalid type."]
                elif isinstance(e, ValueRangeError):
                    if e.min_value is not None and e.max_value is not None:
                        return [f"The {e.field} must be between {e.min_value} and {e.max_value}."]
                    elif e.min_value is not None:
                        return [f"The {e.field} must be at least {e.min_value}."]
                    else:
                        return [f"The {e.field} must be at most {e.max_value}."]
                elif isinstance(e, PatternMismatchError):
                    return [f"The {e.field} has an invalid format."]
                else:
                    return [str(e)]
        
        # Function to demonstrate data validation
        def demonstrate_data_validation():
            # Test with valid data
            valid_data = {
                'username': 'johndoe',
                'email': 'john@example.com',
                'age': 35
            }
            
            # Test with invalid data
            invalid_data_samples = [
                # Missing required field
                {
                    'username': 'janedoe',
                    'age': 28
                    # missing email
                },
                # Type mismatch
                {
                    'username': 'bobsmith',
                    'email': 'bob@example.com',
                    'age': '29'  # age should be an integer
                },
                # Value range error
                {
                    'username': 'alicejones',
                    'email': 'alice@example.com',
                    'age': 17  # too young
                },
                # Pattern mismatch
                {
                    'username': 'charliebrown',
                    'email': 'invalid-email',
                    'age': 42
                }
            ]
            
            # Validate and show errors
            print("Validating valid data:")
            errors = get_validation_errors(valid_data)
            if not errors:
                print("✓ Valid data")
            else:
                print(f"✗ Unexpected errors: {errors}")
            
            print("\nValidating invalid data samples:")
            for i, data in enumerate(invalid_data_samples, 1):
                print(f"\nSample {i}:")
                print(f"Data: {data}")
                
                try:
                    validate_user_data(data)
                    print("✓ Unexpectedly valid")
                except ValidationError as e:
                    print(f"✗ Exception: {e} (type: {type(e).__name__})")
                    print(f"  Details: {e.as_dict()}")
                    
                    # Get user-friendly messages
                    messages = get_validation_errors(data)
                    print(f"  User message: {messages[0]}")
        
        # Run the demonstration
        demonstrate_data_validation()
                        

Exception Handling in Async Functions

        # File: advanced/async_exceptions.py
        
        import asyncio
        import random
        import time
        
        class ServiceUnavailableError(Exception):
            """Raised when a service is unavailable."""
            pass
        
        class TimeoutError(Exception):
            """Raised when an operation times out."""
            pass
        
        async def fetch_data(service_id):
            """
            Simulate fetching data from a service.
            
            Args:
                service_id: ID of the service to fetch from
            
            Returns:
                The fetched data
                
            Raises:
                ServiceUnavailableError: If the service is unavailable
                TimeoutError: If the operation times out
            """
            # Simulate random service behavior
            service_status = random.choice(['available', 'unavailable', 'slow'])
            
            if service_status == 'unavailable':
                raise ServiceUnavailableError(f"Service {service_id} is unavailable")
            
            # Simulate a delay
            delay = 0.1 if service_status == 'available' else 2.0
            await asyncio.sleep(delay)
            
            # Simulate timeout for slow services
            if service_status == 'slow':
                raise TimeoutError(f"Request to service {service_id} timed out")
            
            # If we got here, return some data
            return f"Data from service {service_id}"
        
        async def fetch_with_retry(service_id, max_retries=3, retry_delay=0.5):
            """
            Fetch data with retry logic.
            
            Args:
                service_id: ID of the service to fetch from
                max_retries: Maximum number of retry attempts
                retry_delay: Delay between retries in seconds
                
            Returns:
                The fetched data
                
            Raises:
                ServiceUnavailableError: If the service is still unavailable after all retries
                TimeoutError: If the operation times out on the final retry
            """
            for attempt in range(1, max_retries + 1):
                try:
                    print(f"Fetching from service {service_id}, attempt {attempt}/{max_retries}")
                    return await fetch_data(service_id)
                except (ServiceUnavailableError, TimeoutError) as e:
                    print(f"Attempt {attempt} failed: {e}")
                    
                    if attempt == max_retries:
                        print(f"All {max_retries} attempts failed for service {service_id}")
                        raise  # Re-raise the last exception
                    
                    # Wait before retrying
                    retry_wait = retry_delay * (2 ** (attempt - 1))  # Exponential backoff
                    print(f"Waiting {retry_wait:.2f}s before retry...")
                    await asyncio.sleep(retry_wait)
        
        async def fetch_from_multiple_services(service_ids):
            """
            Fetch data from multiple services concurrently.
            
            Args:
                service_ids: List of service IDs to fetch from
                
            Returns:
                Dictionary mapping service IDs to their data or error messages
            """
            tasks = {}
            results = {}
            
            # Start all fetch tasks
            for service_id in service_ids:
                task = asyncio.create_task(fetch_with_retry(service_id))
                tasks[service_id] = task
            
            # Wait for all tasks to complete (whether successfully or with exception)
            for service_id, task in tasks.items():
                try:
                    data = await task
                    results[service_id] = {'status': 'success', 'data': data}
                except Exception as e:
                    results[service_id] = {'status': 'error', 'error': str(e)}
            
            return results
        
        async def process_in_order(service_ids):
            """
            Process services in order, handling exceptions for each.
            
            Args:
                service_ids: List of service IDs to fetch from
                
            Returns:
                List of results in the same order as service_ids
            """
            results = []
            
            for service_id in service_ids:
                try:
                    data = await fetch_with_retry(service_id)
                    results.append({'service_id': service_id, 'status': 'success', 'data': data})
                except Exception as e:
                    results.append({'service_id': service_id, 'status': 'error', 'error': str(e)})
            
            return results
        
        async def demonstrate_async_exception_handling():
            # Example 1: Basic async exception handling
            print("Example 1: Basic async exception handling")
            try:
                data = await fetch_data(1)
                print(f"Fetched data: {data}")
            except ServiceUnavailableError as e:
                print(f"Service error: {e}")
            except TimeoutError as e:
                print(f"Timeout error: {e}")
            except Exception as e:
                print(f"Unexpected error: {e}")
            
            # Example 2: Retry pattern
            print("\nExample 2: Retry pattern")
            try:
                data = await fetch_with_retry(2, max_retries=3)
                print(f"Fetched data with retry: {data}")
            except Exception as e:
                print(f"All retries failed: {e}")
            
            # Example 3: Concurrent tasks with individual exception handling
            print("\nExample 3: Concurrent tasks")
            service_ids = [3, 4, 5, 6, 7]
            results = await fetch_from_multiple_services(service_ids)
            
            print("Results from concurrent fetches:")
            for service_id, result in results.items():
                if result['status'] == 'success':
                    print(f"Service {service_id}: {result['data']}")
                else:
                    print(f"Service {service_id}: Error - {result['error']}")
            
            # Example 4: Sequential processing with individual exception handling
            print("\nExample 4: Sequential processing")
            service_ids = [8, 9, 10]
            results = await process_in_order(service_ids)
            
            print("Results from sequential processing:")
            for result in results:
                service_id = result['service_id']
                if result['status'] == 'success':
                    print(f"Service {service_id}: {result['data']}")
                else:
                    print(f"Service {service_id}: Error - {result['error']}")
        
        # Run the demonstration
        async def main():
            await demonstrate_async_exception_handling()
        
        # This code would normally be run like:
        # asyncio.run(main())
                        

Global Exception Handler

        # File: patterns/global_exception_handler.py
        
        import sys
        import traceback
        import logging
        from datetime import datetime
        import json
        
        # Configure logging
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler("app.log"),
                logging.StreamHandler()
            ]
        )
        
        logger = logging.getLogger("ExceptionHandler")
        
        class GlobalExceptionHandler:
            """
            A class to handle uncaught exceptions globally.
            """
            
            def __init__(self, log_file=None, report_url=None):
                self.log_file = log_file
                self.report_url = report_url
                self.original_excepthook = sys.excepthook
            
            def handle_exception(self, exc_type, exc_value, exc_traceback):
                """Handle an uncaught exception."""
                # Don't handle KeyboardInterrupt (Ctrl+C)
                if issubclass(exc_type, KeyboardInterrupt):
                    return self.original_excepthook(exc_type, exc_value, exc_traceback)
                
                # Log the exception
                logger.error(
                    "Uncaught exception",
                    exc_info=(exc_type, exc_value, exc_traceback)
                )
                
                # Get formatted traceback
                tb_lines = traceback.format_exception(exc_type, exc_value, exc_traceback)
                tb_text = ''.join(tb_lines)
                
                # Create error report
                error_report = {
                    'timestamp': datetime.now().isoformat(),
                    'exception_type': exc_type.__name__,
                    'exception_message': str(exc_value),
                    'traceback': tb_text
                }
                
                # Write error report to file if configured
                if self.log_file:
                    try:
                        with open(self.log_file, 'a') as f:
                            f.write(f"\n\n--- Error Report: {error_report['timestamp']} ---\n")
                            f.write(json.dumps(error_report, indent=2))
                            f.write("\n------------------------------\n")
                    except Exception as e:
                        logger.error(f"Failed to write error report to file: {e}")
                
                # Send error report to remote service if configured
                if self.report_url:
                    try:
                        # This would normally use requests or similar
                        # requests.post(self.report_url, json=error_report)
                        logger.info(f"Would send error report to {self.report_url}")
                    except Exception as e:
                        logger.error(f"Failed to send error report: {e}")
                
                # Display a user-friendly message
                print("\n===== Unexpected Error =====")
                print(f"An unexpected error has occurred: {exc_type.__name__}")
                print(f"Error details: {exc_value}")
                print("\nThis error has been logged and reported to our team.")
                print("Please check the application logs for more information.")
                print("=============================\n")
            
            def install(self):
                """Install this handler as the global exception handler."""
                sys.excepthook = self.handle_exception
                logger.info("Global exception handler installed")
            
            def uninstall(self):
                """Restore the original exception handler."""
                sys.excepthook = self.original_excepthook
                logger.info("Global exception handler uninstalled")
        
        # Example usage
        def demonstrate_global_handler():
            # Install global handler
            handler = GlobalExceptionHandler(
                log_file="error_reports.log",
                report_url="https://api.example.com/error-reports"
            )
            handler.install()
            
            # Simulate some operations
            try:
                # This will be caught by the try/except block
                result = 1 / 0
            except ZeroDivisionError:
                print("Caught division by zero error")
            
            # This would normally be uncaught and handled by the global handler
            # Uncomment to test
            # x = undefined_variable  # NameError
            
            # Uninstall handler
            handler.uninstall()
        
        # Run the demonstration
        # demonstrate_global_handler()
                        

Error Handling Middleware

        # File: patterns/error_handling_middleware.py
        
        # Example with a web framework (simplified)
        class Request:
            """Simplified request object."""
            def __init__(self, path, method="GET", data=None, headers=None):
                self.path = path
                self.method = method
                self.data = data or {}
                self.headers = headers or {}
        
        class Response:
            """Simplified response object."""
            def __init__(self, body="", status=200, headers=None):
                self.body = body
                self.status = status
                self.headers = headers or {}
            
            def set_status(self, status):
                self.status = status
                return self
            
            def set_body(self, body):
                self.body = body
                return self
            
            def set_header(self, name, value):
                self.headers[name] = value
                return self
            
            def __str__(self):
                return f"Response(status={self.status}, body={self.body!r})"
        
        class ErrorHandlingMiddleware:
            """
            Middleware that catches exceptions from route handlers and converts
            them to appropriate HTTP responses.
            """
            
            def __init__(self, app, logger=None):
                self.app = app
                self.logger = logger
            
            def __call__(self, request):
                try:
                    # Call the next middleware or route handler
                    return self.app(request)
                except Exception as e:
                    # Log the exception
                    if self.logger:
                        self.logger.error(f"Error handling request: {e}", exc_info=True)
                    
                    # Convert exception to appropriate response
                    return self.exception_to_response(e)
            
            def exception_to_response(self, exception):
                """Convert an exception to an HTTP response."""
                # Create a basic error response
                response = Response().set_header("Content-Type", "application/json")
                
                # Customize based on exception type
                if isinstance(exception, NotFoundError):
                    return response.set_status(404).set_body(
                        '{"error": "Not found", "message": "' + str(exception) + '"}'
                    )
                
                elif isinstance(exception, ValidationError):
                    return response.set_status(400).set_body(
                        '{"error": "Bad request", "message": "' + str(exception) + '"}'
                    )
                
                elif isinstance(exception, AuthenticationError):
                    return response.set_status(401).set_body(
                        '{"error": "Unauthorized", "message": "' + str(exception) + '"}'
                    )
                
                elif isinstance(exception, PermissionError):
                    return response.set_status(403).set_body(
                        '{"error": "Forbidden", "message": "' + str(exception) + '"}'
                    )
                
                # Default to 500 Internal Server Error
                else:
                    return response.set_status(500).set_body(
                        '{"error": "Internal server error", "message": "An unexpected error occurred"}'
                    )
        
        # Example custom exceptions for a web application
        class NotFoundError(Exception):
            """Raised when a requested resource doesn't exist."""
            pass
        
        class ValidationError(Exception):
            """Raised when request data fails validation."""
            pass
        
        class AuthenticationError(Exception):
            """Raised when authentication fails."""
            pass
        
        # Example route handlers
        def get_user(request):
            """Example route handler to get a user."""
            user_id = request.path.split("/")[-1]
            
            # Simulate looking up a user
            if user_id == "123":
                return Response().set_body('{"id": "123", "name": "Alice"}')
            else:
                raise NotFoundError(f"User with ID {user_id} not found")
        
        def create_user(request):
            """Example route handler to create a user."""
            # Validate request data
            if 'name' not in request.data:
                raise ValidationError("Name is required")
            
            if not isinstance(request.data.get('name'), str):
                raise ValidationError("Name must be a string")
            
            # Simulate creating a user
            return Response().set_status(201).set_body(
                '{"id": "456", "name": "' + request.data['name'] + '"}'
            )
        
        def authenticate(request):
            """Example route handler for authentication."""
            # Validate credentials
            if 'username' not in request.data or 'password' not in request.data:
                raise ValidationError("Username and password are required")
            
            # Simulate authentication
            if request.data['username'] == "admin" and request.data['password'] == "secret":
                return Response().set_body('{"token": "abc123"}')
            else:
                raise AuthenticationError("Invalid credentials")
        
        # Router with middleware
        class Router:
            """Simplified router with middleware support."""
            
            def __init__(self):
                self.routes = {}
                self.middleware = []
            
            def add_route(self, path, method, handler):
                """Add a route."""
                key = (path, method.upper())
                self.routes[key] = handler
            
            def add_middleware(self, middleware):
                """Add middleware."""
                self.middleware.append(middleware)
            
            def __call__(self, request):
                """Handle a request."""
                # Build the middleware chain
                handler = self.get_handler(request)
                
                # Apply middleware in reverse order
                for middleware in reversed(self.middleware):
                    handler = middleware(handler)
                
                # Handle the request
                return handler(request)
            
            def get_handler(self, request):
                """Get the handler for a request."""
                key = (request.path, request.method.upper())
                
                # Find an exact match
                if key in self.routes:
                    return self.routes[key]
                
                # Find a path pattern match (simplified)
                for (path, method), handler in self.routes.items():
                    if method == request.method.upper() and path.endswith("/:id") and request.path.startswith(path[:-4]):
                        return handler
                
                # No matching route
                return lambda r: Response().set_status(404).set_body(
                    '{"error": "Not found", "message": "Route not found"}'
                )
        
        def demonstrate_middleware():
            # Create a router
            router = Router()
            
            # Add routes
            router.add_route("/users/:id", "GET", get_user)
            router.add_route("/users", "POST", create_user)
            router.add_route("/auth", "POST", authenticate)
            
            # Add middleware
            logger = logging.getLogger("WebApp")
            router.add_middleware(lambda app: ErrorHandlingMiddleware(app, logger))
            
            # Test requests
            requests = [
                # Valid request, should succeed
                Request("/users/123", "GET"),
                
                # Invalid user ID, should return 404
                Request("/users/999", "GET"),
                
                # Valid user creation
                Request("/users", "POST", {"name": "Bob"}),
                
                # Invalid user creation (missing name)
                Request("/users", "POST", {}),
                
                # Valid authentication
                Request("/auth", "POST", {"username": "admin", "password": "secret"}),
                
                # Invalid authentication
                Request("/auth", "POST", {"username": "admin", "password": "wrong"})
            ]
            
            # Process each request
            for i, request in enumerate(requests, 1):
                print(f"\nRequest {i}: {request.method} {request.path}")
                if request.data:
                    print(f"Data: {request.data}")
                
                response = router(request)
                print(f"Response: {response}")
        
        # Run the demonstration
        # demonstrate_middleware()
                        

Avoid Using Exceptions for Flow Control

        # File: performance/exception_vs_check.py
        
        import time
        import statistics
        
        # Example 1: Using exceptions for control flow
        def find_key_with_exception(data, key):
            try:
                return data[key]
            except KeyError:
                return None
        
        # Example 2: Using a check first
        def find_key_with_check(data, key):
            if key in data:
                return data[key]
            else:
                return None
        
        # Timing function
        def time_function(func, *args, **kwargs):
            """Measure the execution time of a function."""
            start_time = time.time()
            result = func(*args, **kwargs)
            end_time = time.time()
            return result, (end_time - start_time) * 1000  # Convert to milliseconds
        
        # Performance comparison
        def compare_approaches(data_size=1000000, iterations=100):
            # Create a large dictionary
            data = {f"key_{i}": i for i in range(data_size)}
            
            # Test keys
            existing_key = "key_1000"  # This key exists
            missing_key = "nonexistent_key"  # This key doesn't exist
            
            # Measure time for existing key with exception approach
            exception_times_existing = []
            for _ in range(iterations):
                _, elapsed = time_function(find_key_with_exception, data, existing_key)
                exception_times_existing.append(elapsed)
            
            # Measure time for missing key with exception approach
            exception_times_missing = []
            for _ in range(iterations):
                _, elapsed = time_function(find_key_with_exception, data, missing_key)
                exception_times_missing.append(elapsed)
            
            # Measure time for existing key with check approach
            check_times_existing = []
            for _ in range(iterations):
                _, elapsed = time_function(find_key_with_check, data, existing_key)
                check_times_existing.append(elapsed)
            
            # Measure time for missing key with check approach
            check_times_missing = []
            for _ in range(iterations):
                _, elapsed = time_function(find_key_with_check, data, missing_key)
                check_times_missing.append(elapsed)
            
            # Print results
            print(f"Performance comparison ({data_size} items, {iterations} iterations):")
            print("\nExisting key:")
            print(f"  Exception approach: {statistics.mean(exception_times_existing):.4f} ms")
            print(f"  Check approach:     {statistics.mean(check_times_existing):.4f} ms")
            
            print("\nMissing key:")
            print(f"  Exception approach: {statistics.mean(exception_times_missing):.4f} ms")
            print(f"  Check approach:     {statistics.mean(check_times_missing):.4f} ms")
        
        # Run the performance comparison
        # compare_approaches()