Advanced Exception Handling: Power Techniques

Welcome to Advanced Exception Handling! Now that you know the basics, let’s explore powerful techniques for handling complex error scenarios. Think of this as upgrading from a basic safety net to a professional error management system.

Raising Exceptions

The raise Statement

Sometimes you need to create your own exceptions when something goes wrong in your code.

def validate_age(age):
    if age < 0:
        raise ValueError("Age cannot be negative")
    if age > 150:
        raise ValueError("Age seems unreasonably high")
    return age

# Test
try:
    validate_age(-5)
except ValueError as e:
    print(f"Validation error: {e}")

try:
    validate_age(200)
except ValueError as e:
    print(f"Validation error: {e}")

Re-raising Exceptions

def process_data(data):
    try:
        # Some processing that might fail
        result = complex_calculation(data)
        return result
    except ValueError as e:
        # Log the error but re-raise it
        print(f"Data processing failed: {e}")
        raise  # Re-raises the same exception

# Usage
try:
    process_data("invalid_data")
except ValueError:
    print("Caller also handles the error")

Custom Exception Classes

Creating Custom Exceptions

class ValidationError(Exception):
    """Custom exception for validation errors."""
    pass

class InsufficientFundsError(Exception):
    """Custom exception for bank account errors."""

    def __init__(self, balance, amount):
        self.balance = balance
        self.amount = amount
        super().__init__(f"Insufficient funds: balance ${balance}, needed ${amount}")

class BankAccount:
    def __init__(self, balance=0):
        self.balance = balance

    def withdraw(self, amount):
        if amount <= 0:
            raise ValidationError("Withdrawal amount must be positive")
        if amount > self.balance:
            raise InsufficientFundsError(self.balance, amount)

        self.balance -= amount
        return self.balance

# Test custom exceptions
account = BankAccount(100)

try:
    account.withdraw(-50)
except ValidationError as e:
    print(f"Validation error: {e}")

try:
    account.withdraw(150)
except InsufficientFundsError as e:
    print(f"Bank error: {e}")
    print(f"Current balance: ${e.balance}")
    print(f"Amount needed: ${e.amount}")

Exception Hierarchy

class ApplicationError(Exception):
    """Base class for all application errors."""
    pass

class DatabaseError(ApplicationError):
    """Database-related errors."""
    pass

class NetworkError(ApplicationError):
    """Network-related errors."""
    pass

class ConnectionError(NetworkError):
    """Connection-specific errors."""
    pass

# Usage
def connect_to_database():
    # Simulate different error conditions
    error_type = "connection"  # Could be "network", "database", etc.

    if error_type == "connection":
        raise ConnectionError("Failed to establish database connection")
    elif error_type == "network":
        raise NetworkError("Network timeout")
    else:
        raise DatabaseError("Unknown database error")

# Exception handling with hierarchy
try:
    connect_to_database()
except ConnectionError as e:
    print(f"Connection failed: {e}")
    # Handle connection-specific recovery
except NetworkError as e:
    print(f"Network issue: {e}")
    # Handle network issues
except DatabaseError as e:
    print(f"Database error: {e}")
    # Handle general database issues
except ApplicationError as e:
    print(f"Application error: {e}")
    # Handle any application error

Exception Chaining

Chaining Exceptions with raise ... from

def read_config(filename):
    try:
        with open(filename, "r") as file:
            return file.read()
    except FileNotFoundError as e:
        raise RuntimeError(f"Configuration file missing: {filename}") from e

def load_configuration():
    try:
        config_data = read_config("config.json")
        # Process config...
        return config_data
    except RuntimeError as e:
        raise ValueError("Failed to load application configuration") from e

# Usage
try:
    load_configuration()
except ValueError as e:
    print(f"Configuration loading failed: {e}")
    print(f"Original cause: {e.__cause__}")  # Access the chained exception

Suppressing Exception Context

def safe_divide(a, b):
    try:
        return a / b
    except ZeroDivisionError:
        # Replace the exception, don't chain it
        raise ValueError("Cannot divide by zero") from None

# The original ZeroDivisionError is not shown in the traceback
try:
    safe_divide(10, 0)
except ValueError as e:
    print(f"Error: {e}")
    # No __cause__ attribute because of 'from None'

Context Managers and Exceptions

Custom Context Managers

class DatabaseConnection:
    def __init__(self, connection_string):
        self.connection_string = connection_string
        self.connection = None

    def __enter__(self):
        # Setup code
        print(f"Connecting to: {self.connection_string}")
        self.connection = "simulated_connection"  # In real code: actual connection
        return self.connection

    def __exit__(self, exc_type, exc_val, exc_tb):
        # Cleanup code - always executed
        print("Closing database connection")
        if self.connection:
            # In real code: self.connection.close()
            self.connection = None

        # Return False to propagate exceptions, True to suppress
        if exc_type is not None:
            print(f"Exception occurred: {exc_val}")
        return False  # Propagate the exception

# Usage
with DatabaseConnection("postgresql://localhost/mydb") as conn:
    print("Performing database operations...")
    # Simulate an error
    raise ValueError("Database operation failed")

print("Connection closed automatically")

Exception Handling in Context Managers

class FileProcessor:
    def __init__(self, filename):
        self.filename = filename
        self.file = None

    def __enter__(self):
        self.file = open(self.filename, "r")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.file:
            self.file.close()

        # Handle specific exceptions
        if exc_type == ValueError:
            print(f"Data processing error in {self.filename}: {exc_val}")
            return True  # Suppress the exception
        elif exc_type == FileNotFoundError:
            print(f"File not found: {self.filename}")
            return True  # Suppress the exception

        return False  # Propagate other exceptions

    def process(self):
        content = self.file.read()
        # Simulate processing that might fail
        if "error" in content.lower():
            raise ValueError("Invalid data found")
        return content.upper()

# Usage
with FileProcessor("data.txt") as processor:
    result = processor.process()
    print(f"Processed: {result}")

# Even if exceptions occur, file is closed

Advanced Try-Except Patterns

Nested Try-Except Blocks

def complex_operation(data):
    try:
        # Outer try block
        print("Starting complex operation...")

        try:
            # Inner try block for specific operation
            result = int(data)
            print(f"Converted to integer: {result}")
        except ValueError:
            print("Inner: Could not convert to integer, trying float...")
            result = float(data)
            print(f"Converted to float: {result}")

        # Continue with result
        final_result = result * 2
        print(f"Final result: {final_result}")
        return final_result

    except Exception as e:
        print(f"Outer: Unexpected error: {e}")
        return None

# Test
complex_operation("42")      # Success with int
complex_operation("3.14")    # Success with float
complex_operation("hello")   # Error

Exception Groups (Python 3.11+)

# Exception groups allow handling multiple exceptions at once
try:
    # Code that might raise multiple exceptions
    raise ExceptionGroup("Multiple errors occurred", [
        ValueError("Invalid input"),
        FileNotFoundError("Config file missing"),
        ConnectionError("Network timeout")
    ])
except* ValueError as e:
    print(f"Validation errors: {e}")
except* (FileNotFoundError, ConnectionError) as e:
    print(f"System errors: {e}")

Logging Exceptions

Basic Exception Logging

import logging
import traceback

# Configure logging
logging.basicConfig(
    filename="error.log",
    level=logging.ERROR,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

def risky_operation():
    try:
        # Some operation that might fail
        result = 10 / 0
        return result
    except Exception as e:
        # Log the exception
        logging.error(f"Error in risky_operation: {e}")
        logging.error(f"Traceback: {traceback.format_exc()}")

        # Also print to console for immediate feedback
        print(f"An error occurred: {e}")
        return None

# Test
risky_operation()

Advanced Logging with Context

import logging
import sys
from datetime import datetime

class ErrorLogger:
    def __init__(self, log_file="errors.log"):
        self.logger = logging.getLogger("ErrorLogger")
        self.logger.setLevel(logging.ERROR)

        # File handler
        file_handler = logging.FileHandler(log_file)
        file_handler.setLevel(logging.ERROR)

        # Console handler
        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setLevel(logging.WARNING)

        # Formatter
        formatter = logging.Formatter(
            "%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s"
        )
        file_handler.setFormatter(formatter)
        console_handler.setFormatter(formatter)

        self.logger.addHandler(file_handler)
        self.logger.addHandler(console_handler)

    def log_exception(self, e, context=None):
        """Log an exception with additional context."""
        error_msg = f"Exception: {type(e).__name__}: {e}"

        if context:
            error_msg += f" | Context: {context}"

        self.logger.error(error_msg)
        self.logger.error(f"Traceback: {traceback.format_exc()}")

# Usage
error_logger = ErrorLogger()

def process_user_data(user_id, data):
    try:
        # Simulate processing
        if not data:
            raise ValueError("Empty data provided")

        result = data.upper()
        return result

    except Exception as e:
        context = {
            "user_id": user_id,
            "data_length": len(data) if data else 0,
            "timestamp": datetime.now().isoformat()
        }
        error_logger.log_exception(e, context)
        return None

# Test
process_user_data(123, "")  # Will log error
process_user_data(456, "valid data")  # Success

Defensive Programming Techniques

Input Validation with Exceptions

def validate_user_input(data):
    """Validate user input with detailed error messages."""
    errors = []

    # Check required fields
    required_fields = ["name", "email", "age"]
    for field in required_fields:
        if field not in data:
            errors.append(f"Missing required field: {field}")

    # Validate name
    if "name" in data:
        name = data["name"]
        if not isinstance(name, str) or len(name.strip()) < 2:
            errors.append("Name must be a string with at least 2 characters")

    # Validate email
    if "email" in data:
        email = data["email"]
        if "@" not in email or "." not in email.split("@")[1]:
            errors.append("Invalid email format")

    # Validate age
    if "age" in data:
        try:
            age = int(data["age"])
            if age < 0 or age > 150:
                errors.append("Age must be between 0 and 150")
        except (ValueError, TypeError):
            errors.append("Age must be a valid number")

    if errors:
        raise ValidationError("Input validation failed", errors)

    return True

class ValidationError(Exception):
    def __init__(self, message, errors):
        self.errors = errors
        super().__init__(message)

# Usage
try:
    user_data = {
        "name": "A",
        "email": "invalid-email",
        "age": "not-a-number"
    }
    validate_user_input(user_data)
except ValidationError as e:
    print(f"Validation failed: {e}")
    for error in e.errors:
        print(f"  - {error}")

Resource Management with Context Managers

from contextlib import contextmanager
import time

@contextmanager
def timed_operation(operation_name):
    """Context manager that times an operation."""
    start_time = time.time()
    print(f"Starting {operation_name}...")
    try:
        yield
    except Exception as e:
        elapsed = time.time() - start_time
        print(f"{operation_name} failed after {elapsed:.2f}s: {e}")
        raise
    else:
        elapsed = time.time() - start_time
        print(f"{operation_name} completed in {elapsed:.2f}s")

@contextmanager
def database_transaction(connection):
    """Context manager for database transactions."""
    try:
        # Begin transaction
        connection.begin_transaction()
        yield connection
        # Commit if no exceptions
        connection.commit()
        print("Transaction committed")
    except Exception as e:
        # Rollback on error
        connection.rollback()
        print(f"Transaction rolled back: {e}")
        raise

# Usage
with timed_operation("data processing"):
    # Simulate some work
    time.sleep(1)
    print("Processing data...")

# Database example (simulated)
class MockConnection:
    def begin_transaction(self): print("Beginning transaction")
    def commit(self): print("Committing transaction")
    def rollback(self): print("Rolling back transaction")

conn = MockConnection()
try:
    with database_transaction(conn):
        print("Performing database operations...")
        # Simulate error
        raise ValueError("Database error")
except ValueError:
    print("Handled database error")

Real-World Examples

Example 1: Robust API Client

import requests
import time
from requests.exceptions import RequestException, Timeout, ConnectionError

class APIClient:
    def __init__(self, base_url, max_retries=3, timeout=10):
        self.base_url = base_url
        self.max_retries = max_retries
        self.timeout = timeout

    def make_request(self, endpoint, method="GET", **kwargs):
        """Make a robust API request with retry logic."""
        url = f"{self.base_url}/{endpoint}"

        for attempt in range(self.max_retries + 1):
            try:
                response = requests.request(
                    method=method,
                    url=url,
                    timeout=self.timeout,
                    **kwargs
                )
                response.raise_for_status()  # Raise for bad status codes
                return response.json()

            except Timeout:
                if attempt == self.max_retries:
                    raise APIError(f"Request timeout after {self.max_retries + 1} attempts")
                print(f"Timeout, retrying... (attempt {attempt + 1})")
                time.sleep(2 ** attempt)  # Exponential backoff

            except ConnectionError:
                if attempt == self.max_retries:
                    raise APIError(f"Connection failed after {self.max_retries + 1} attempts")
                print(f"Connection error, retrying... (attempt {attempt + 1})")
                time.sleep(2 ** attempt)

            except requests.exceptions.HTTPError as e:
                # Don't retry for HTTP errors (4xx, 5xx)
                raise APIError(f"HTTP error: {e.response.status_code} - {e.response.text}")

            except RequestException as e:
                if attempt == self.max_retries:
                    raise APIError(f"Request failed: {e}")
                print(f"Request error, retrying... (attempt {attempt + 1})")
                time.sleep(2 ** attempt)

class APIError(Exception):
    """Custom exception for API errors."""
    pass

# Usage
client = APIClient("https://jsonplaceholder.typicode.com")

try:
    # This will work
    data = client.make_request("posts/1")
    print(f"Retrieved post: {data['title']}")

    # This will fail gracefully
    data = client.make_request("nonexistent/endpoint")

except APIError as e:
    print(f"API request failed: {e}")

Example 2: File Processing Pipeline

import os
import json
from pathlib import Path

class DataProcessingError(Exception):
    """Custom exception for data processing errors."""
    pass

class FileProcessingPipeline:
    def __init__(self, input_dir, output_dir):
        self.input_dir = Path(input_dir)
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(exist_ok=True)

    def process_files(self):
        """Process all files in the input directory."""
        processed_count = 0
        errors = []

        for file_path in self.input_dir.glob("*.json"):
            try:
                with self.process_file(file_path) as result:
                    output_file = self.output_dir / f"processed_{file_path.name}"
                    with open(output_file, "w") as f:
                        json.dump(result, f, indent=2)
                    processed_count += 1

            except DataProcessingError as e:
                errors.append(f"{file_path.name}: {e}")
                continue
            except Exception as e:
                errors.append(f"{file_path.name}: Unexpected error - {e}")
                continue

        return processed_count, errors

    @contextmanager
    def process_file(self, file_path):
        """Context manager for processing individual files."""
        try:
            with open(file_path, "r") as f:
                data = json.load(f)

            # Validate data structure
            if not isinstance(data, dict):
                raise DataProcessingError("Data must be a JSON object")

            if "records" not in data:
                raise DataProcessingError("Missing 'records' field")

            # Process the data
            processed_data = {
                "filename": file_path.name,
                "record_count": len(data["records"]),
                "processed_at": str(datetime.now()),
                "summary": self._calculate_summary(data["records"])
            }

            yield processed_data

        except json.JSONDecodeError as e:
            raise DataProcessingError(f"Invalid JSON format: {e}")
        except FileNotFoundError:
            raise DataProcessingError("File not found")
        except PermissionError:
            raise DataProcessingError("Permission denied")

    def _calculate_summary(self, records):
        """Calculate summary statistics."""
        if not records:
            return {"total": 0, "average": 0}

        total = sum(record.get("value", 0) for record in records)
        average = total / len(records)
        return {"total": total, "average": round(average, 2)}

# Usage
pipeline = FileProcessingPipeline("input_data", "output_data")
processed, errors = pipeline.process_files()

print(f"Successfully processed {processed} files")
if errors:
    print("Errors encountered:")
    for error in errors:
        print(f"  - {error}")

Best Practices

1. Use Specific Exception Types

# ✅ Good - specific exceptions
def divide_numbers(a, b):
    if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):
        raise TypeError("Both arguments must be numbers")
    if b == 0:
        raise ZeroDivisionError("Cannot divide by zero")
    return a / b

# ❌ Bad - generic exceptions
def divide_numbers(a, b):
    if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):
        raise Exception("Invalid input")
    if b == 0:
        raise Exception("Division error")
    return a / b

2. Create Custom Exception Hierarchies

class ApplicationError(Exception):
    """Base exception for application errors."""
    pass

class ValidationError(ApplicationError):
    """Validation-related errors."""
    pass

class ProcessingError(ApplicationError):
    """Data processing errors."""
    pass

class NetworkError(ApplicationError):
    """Network-related errors."""
    pass

3. Use Context Managers for Resources

# ✅ Good - automatic cleanup
with open("file.txt", "r") as f:
    data = f.read()

# ✅ Good - custom context manager
class DatabaseConnection:
    def __enter__(self):
        # Setup
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        # Cleanup
        pass

4. Log Exceptions Properly

import logging
import traceback

def handle_error(e, context=None):
    logging.error(f"Error: {e}")
    logging.error(f"Traceback: {traceback.format_exc()}")
    if context:
        logging.error(f"Context: {context}")

5. Don’t Overuse Bare Except

# ❌ Bad - catches everything
try:
    risky_operation()
except:
    pass

# ✅ Good - specific handling
try:
    risky_operation()
except ValueError:
    handle_value_error()
except Exception as e:
    logging.error(f"Unexpected error: {e}")
    raise

Practice Exercises

Exercise 1: Custom Exception Classes

Create a custom exception hierarchy for a banking system:

• BankingError (base class)
• InsufficientFundsError
• InvalidAccountError
• TransactionError

Implement a BankAccount class that raises these exceptions appropriately.

Exercise 2: Retry Decorator

Create a decorator that automatically retries functions on failure:

• Configurable number of retries
• Exponential backoff
• Customizable exceptions to retry
• Logging of retry attempts

Exercise 3: Context Manager for Timing

Build a context manager that:

• Times code execution
• Logs slow operations
• Handles exceptions during timing
• Provides execution statistics

Exercise 4: Validation Framework

Create a validation framework that:

• Validates data against schemas
• Raises detailed validation errors
• Supports nested validation
• Provides error aggregation

Exercise 5: Error Recovery System

Build an error recovery system that:

• Attempts multiple recovery strategies
• Logs recovery attempts
• Escalates unrecoverable errors
• Provides recovery statistics

Summary

Advanced exception handling takes your error management to the next level:

Raising Exceptions:

raise ValueError("Invalid input")
raise CustomError("Something went wrong") from original_exception

Custom Exceptions:

class MyError(Exception):
    def __init__(self, message, code=None):
        self.code = code
        super().__init__(message)

Context Managers:

class MyContext:
    def __enter__(self): return self
    def __exit__(self, exc_type, exc_val, exc_tb): pass

Exception Chaining:

• Use raise ... from e to preserve error context
• Use from None to replace exception context

Best Practices:

• Create exception hierarchies
• Use context managers for resources
• Log exceptions with context
• Implement retry logic for transient errors
• Validate input thoroughly

Next: Defensive Programming - preventing errors before they happen! 🛡️