🚨 Exceptions Overview — Advanced Patterns and Performance

Explore exception mechanics at a deeper level, including performance considerations and production patterns.

🔬 Exception Creation and Cost Analysis

Creating exceptions is expensive—understand the performance implications:

import time
import sys

# Measuring exception creation overhead
def measure_exception_creation():
    # Exception creation includes traceback capture
    start = time.perf_counter()
    for _ in range(10000):
        try:
            raise ValueError("test")
        except ValueError:
            pass
    exception_time = time.perf_counter() - start

    # Compare with normal operations
    start = time.perf_counter()
    for _ in range(10000):
        result = int("123")
    normal_time = time.perf_counter() - start

    print(f"Exception time: {exception_time:.4f}s")
    print(f"Normal time: {normal_time:.4f}s")
    print(f"Ratio: {exception_time / normal_time:.1f}x slower")

# Result: Exceptions are ~100-1000x slower than normal operations
measure_exception_creation()

# Never use exceptions for control flow!
# ❌ BAD: Using exceptions for control flow
def find_index_bad(items, value):
    try:
        return items.index(value)
    except ValueError:
        return -1

# ✅ GOOD: Use normal control flow
def find_index_good(items, value):
    try:
        idx = items.index(value)
        return idx
    except ValueError:
        return -1

# Even better: don't use exceptions at all for this
def find_index_best(items, value):
    if value in items:
        return items.index(value)
    return -1

🔗 Exception Chaining and Context

Understanding how exceptions relate to each other:

# Implicit exception chaining
def load_user_from_api(user_id):
    try:
        response = requests.get(f"https://api.example.com/users/{user_id}")
        data = response.json()
        return data
    except requests.ConnectionError as e:
        # When you raise here, Python remembers the original exception
        raise ValueError(f"Cannot load user {user_id}") from e

# Try it
try:
    user = load_user_from_api(123)
except ValueError as e:
    # e.__cause__ points to the original ConnectionError
    print(f"Original cause: {e.__cause__}")
    print(f"Traceback shows both exceptions!")

# Explicit exception chaining - preserving context
import json

def parse_user_json(json_string):
    try:
        data = json.loads(json_string)
    except json.JSONDecodeError as e:
        # 'from e' creates exception chain
        raise ValueError("Invalid user JSON") from e

# Suppress exception context if needed (rare)
def operation_with_cleanup():
    try:
        risky_operation()
    except Exception as original:
        try:
            cleanup()
        except Exception:
            pass  # Don't care about cleanup exception
        raise original  # Re-raise original without modification

# Exception context attributes
try:
    try:
        x = 1 / 0
    except ZeroDivisionError as e:
        raise ValueError("Math error") from e
except ValueError as e:
    print(f"Exception: {e}")
    print(f"Cause (__cause__): {e.__cause__}")
    print(f"Context (__context__): {e.__context__}")
    print(f"Has explicit cause: {e.__cause__ is not None}")

🎯 Custom Exception Hierarchies for Large Systems

# Well-designed exception hierarchy for complex application
class ApplicationError(Exception):
    """Base exception for entire application"""
    pass

class ConfigurationError(ApplicationError):
    """Configuration-related errors"""
    pass

class DataError(ApplicationError):
    """Data processing errors"""
    pass

class ValidationError(DataError):
    """User input validation failures"""
    def __init__(self, field, message, value=None):
        self.field = field
        self.message = message
        self.value = value
        super().__init__(f"{field}: {message}")

class DataFormatError(DataError):
    """Data format is invalid"""
    pass

class APIError(ApplicationError):
    """External API errors"""
    pass

class APITimeoutError(APIError):
    """API request timeout"""
    pass

class APIRateLimitError(APIError):
    """API rate limit exceeded"""
    def __init__(self, reset_time):
        self.reset_time = reset_time
        super().__init__(f"Rate limited. Reset at {reset_time}")

# Usage patterns
def validate_age(age):
    if not isinstance(age, int):
        raise ValidationError("age", "must be integer", age)
    if age < 0:
        raise ValidationError("age", "cannot be negative", age)

def call_api_with_retry(endpoint, max_retries=3):
    for attempt in range(max_retries):
        try:
            return api_request(endpoint)
        except APITimeoutError as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)
        except APIRateLimitError as e:
            print(f"Rate limited until {e.reset_time}")
            raise
        except APIError:
            raise

# Catching by hierarchy
try:
    validate_age("invalid")
except ValidationError as e:
    print(f"Validation error: {e.field} - {e.message}")
except DataError as e:
    print(f"Data error: {e}")
except ApplicationError as e:
    print(f"Application error: {e}")

📊 Exception Propagation and Stack Unwinding

# Understanding how exceptions propagate
class Transaction:
    def __init__(self, name):
        self.name = name
        print(f"[{self.name}] START")

    def rollback(self):
        print(f"[{self.name}] ROLLBACK")

    def commit(self):
        print(f"[{self.name}] COMMIT")

def level_3():
    print("[Level 3] Executing")
    raise ValueError("Something failed at level 3")
    print("[Level 3] Cleanup (never reached)")

def level_2():
    txn = Transaction("Level2")
    try:
        level_3()
    except ValueError:
        txn.rollback()
        raise  # Re-raise

def level_1():
    txn = Transaction("Level1")
    try:
        level_2()
    except ValueError as e:
        txn.rollback()
        # Don't re-raise - handle here
        print(f"Handled at Level 1: {e}")

# Call stack unwinding
level_1()

# Output shows rollback order (LIFO - Last In First Out)
# [Level1] START
# [Level2] START
# [Level 3] Executing
# [Level2] ROLLBACK
# [Level1] ROLLBACK
# Handled at Level 1: Something failed at level 3

🛡️ Exception Handling Patterns for Distributed Systems

# Pattern 1: Retry with exponential backoff
import random
import time

def call_with_backoff(func, max_retries=5):
    for attempt in range(max_retries):
        try:
            return func()
        except ConnectionError as e:
            if attempt == max_retries - 1:
                raise
            # Exponential backoff with jitter
            wait_time = (2 ** attempt) + random.uniform(0, 1)
            print(f"Attempt {attempt + 1} failed. Retrying in {wait_time:.2f}s")
            time.sleep(wait_time)
        except ValueError:
            # Don't retry for validation errors
            raise

# Pattern 2: Circuit breaker
class CircuitBreaker:
    """Prevents cascading failures"""
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failures = 0
        self.last_failure_time = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN

    def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "HALF_OPEN"
            else:
                raise RuntimeError("Circuit breaker is OPEN")

        try:
            result = func(*args, **kwargs)
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure_time = time.time()
            if self.failures >= self.failure_threshold:
                self.state = "OPEN"
            raise

# Usage
breaker = CircuitBreaker()

def unreliable_api_call():
    # Sometimes fails
    if random.random() < 0.7:
        raise ConnectionError("API unavailable")
    return "success"

for i in range(10):
    try:
        result = breaker.call(unreliable_api_call)
        print(f"Call {i}: {result}")
    except Exception as e:
        print(f"Call {i}: Failed - {type(e).__name__}")

# Pattern 3: Bulkhead isolation
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout

class BulkheadExecutor:
    """Isolates failures to specific component"""
    def __init__(self, max_workers=5, timeout=10):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.timeout = timeout

    def execute(self, func, *args, **kwargs):
        try:
            future = self.executor.submit(func, *args, **kwargs)
            return future.result(timeout=self.timeout)
        except FutureTimeout:
            raise TimeoutError(f"Operation exceeded {self.timeout}s timeout")
        except Exception:
            raise

🔍 Advanced Traceback Analysis

import traceback
import sys

# Detailed traceback information
def deep_function():
    local_var = "important data"
    raise ValueError("Something went wrong")

def middle_function():
    middle_var = 42
    deep_function()

def outer_function():
    outer_var = [1, 2, 3]
    middle_function()

# Capture detailed traceback
try:
    outer_function()
except ValueError as e:
    # Get full exception info
    exc_type, exc_value, exc_traceback = sys.exc_info()

    # Get traceback frames
    tb = traceback.extract_tb(exc_traceback)
    for frame in tb:
        print(f"File: {frame.filename}")
        print(f"Function: {frame.name}")
        print(f"Line: {frame.lineno}")
        print(f"Code: {frame.line}")
        print()

    # Get local variables from each frame
    print("Local variables at crash point:")
    tb = exc_traceback
    while tb:
        frame = tb.tb_frame
        print(f"In function {frame.f_code.co_name}:")
        for var_name, var_value in frame.f_locals.items():
            print(f"  {var_name} = {var_value}")
        tb = tb.tb_next

⚡ Exception Performance Optimization

# Bad: Creating exceptions in hot loop
def process_items_slow(items):
    result = []
    for item in items:
        try:
            processed = process_item(item)
            result.append(processed)
        except ValueError:
            pass  # Ignore errors
    return result

# Better: Check before trying
def process_items_fast(items):
    result = []
    for item in items:
        if is_valid_item(item):
            processed = process_item(item)
            result.append(processed)
    return result

# Or use EAFP (Easier to Ask for Forgiveness than Permission) properly
def process_items_eafp(items):
    result = []
    for item in items:
        try:
            processed = process_item(item)
            result.append(processed)
        except ValueError:
            pass
    return result

# Measure the difference
import timeit

items = list(range(1000))

# These have different performance characteristics depending on error frequency
# If errors are rare, EAFP is faster
# If errors are common, checking first is faster

🎯 Key Takeaways

• ✅ Exceptions are expensive—avoid in hot loops

• ✅ Never use exceptions for normal control flow

• ✅ Design hierarchies for complex applications

• ✅ Use exception chaining with `from e` for context

• ✅ Implement retry patterns with exponential backoff

• ✅ Consider circuit breaker for distributed systems

• ✅ Understand stack unwinding and propagation