Article 9: Error Handling and Logging

Introduction

Errors are inevitable in software—what matters is how we handle them. Through debugging production systems at 3 AM, I've learned that proper error handling and logging are what make the difference between a quick fix and hours of frustration.

This article covers exception handling best practices, structured logging, and debugging strategies that help you build resilient applications.

Exception Handling Fundamentals

The Basics

# Basic try-except
try:
    result = risky_operation()
except SomeException as e:
    handle_error(e)

Exception Hierarchy

Catching Specific Exceptions

# Bad: Catching everything hides bugs
try:
    process_data(data)
except:  # Never do this!
    pass

# Bad: Too broad
try:
    process_data(data)
except Exception:  # Still too broad
    log.error("Something went wrong")

# Good: Catch specific exceptions
try:
    user = fetch_user(user_id)
    process_order(user, order_data)
except UserNotFoundError:
    return {"error": "User not found"}, 404
except InvalidOrderError as e:
    return {"error": str(e)}, 400
except DatabaseConnectionError:
    log.error("Database connection failed")
    return {"error": "Service temporarily unavailable"}, 503

Multiple Exception Types

# Catching multiple exceptions
try:
    value = int(user_input)
    result = data[value]
except (ValueError, IndexError) as e:
    print(f"Invalid input: {e}")

# Different handlers for different exceptions
try:
    config = load_config(path)
except FileNotFoundError:
    config = create_default_config()
except json.JSONDecodeError as e:
    raise ConfigurationError(f"Invalid config file: {e}")
except PermissionError:
    raise ConfigurationError(f"Cannot read config file: {path}")

Custom Exceptions

Creating Custom Exceptions

# Base exception for your application
class TaskManagerError(Exception):
    """Base exception for Task Manager application."""
    pass


# Specific exceptions
class TaskNotFoundError(TaskManagerError):
    """Raised when a task cannot be found."""
    
    def __init__(self, task_id: int) -> None:
        self.task_id = task_id
        super().__init__(f"Task with ID {task_id} not found")


class ValidationError(TaskManagerError):
    """Raised when validation fails."""
    
    def __init__(self, field: str, message: str) -> None:
        self.field = field
        self.message = message
        super().__init__(f"Validation error on '{field}': {message}")


class DuplicateTaskError(TaskManagerError):
    """Raised when attempting to create a duplicate task."""
    
    def __init__(self, title: str) -> None:
        self.title = title
        super().__init__(f"Task with title '{title}' already exists")

Using Custom Exceptions

class TaskService:
    def get_task(self, task_id: int) -> Task:
        """Get a task by ID.
        
        Raises:
            TaskNotFoundError: If task doesn't exist.
        """
        task = self._repository.find_by_id(task_id)
        if task is None:
            raise TaskNotFoundError(task_id)
        return task
    
    def create_task(self, title: str) -> Task:
        """Create a new task.
        
        Raises:
            ValidationError: If title is invalid.
            DuplicateTaskError: If task with title exists.
        """
        if not title or not title.strip():
            raise ValidationError("title", "Title cannot be empty")
        
        if len(title) > 200:
            raise ValidationError("title", "Title cannot exceed 200 characters")
        
        existing = self._repository.find_by_title(title)
        if existing:
            raise DuplicateTaskError(title)
        
        return self._repository.save(Task(title=title))

Exception Context

# Chaining exceptions to preserve context
class DataProcessingError(Exception):
    """Error during data processing."""
    pass


def process_file(path: str) -> dict:
    try:
        with open(path) as f:
            data = json.load(f)
    except FileNotFoundError:
        raise DataProcessingError(f"File not found: {path}") from None
    except json.JSONDecodeError as e:
        raise DataProcessingError(f"Invalid JSON in {path}") from e
    
    try:
        return transform_data(data)
    except KeyError as e:
        raise DataProcessingError(f"Missing required field: {e}") from e

Error Handling Patterns

The Result Pattern

from dataclasses import dataclass
from typing import Generic, TypeVar, Union

T = TypeVar("T")
E = TypeVar("E")


@dataclass
class Success(Generic[T]):
    value: T


@dataclass
class Failure(Generic[E]):
    error: E


Result = Union[Success[T], Failure[E]]


def divide(a: float, b: float) -> Result[float, str]:
    if b == 0:
        return Failure("Cannot divide by zero")
    return Success(a / b)


# Usage
result = divide(10, 2)
match result:
    case Success(value):
        print(f"Result: {value}")
    case Failure(error):
        print(f"Error: {error}")

Fail Fast

def process_order(order: Order) -> Receipt:
    """Process an order - fail fast on invalid input."""
    # Validate all inputs first
    if not order.items:
        raise ValueError("Order must have at least one item")
    
    if order.customer is None:
        raise ValueError("Order must have a customer")
    
    if not order.customer.email:
        raise ValueError("Customer must have an email")
    
    # All validations passed, proceed with processing
    return _process_valid_order(order)

Graceful Degradation

def get_user_profile(user_id: int) -> UserProfile:
    """Get user profile with graceful degradation."""
    user = user_repository.get(user_id)
    
    # Try to get avatar, fall back to default
    try:
        avatar_url = avatar_service.get_url(user_id)
    except AvatarServiceError:
        log.warning(f"Avatar service unavailable for user {user_id}")
        avatar_url = DEFAULT_AVATAR_URL
    
    # Try to get recent activity, fall back to empty
    try:
        recent_activity = activity_service.get_recent(user_id)
    except ActivityServiceError:
        log.warning(f"Activity service unavailable for user {user_id}")
        recent_activity = []
    
    return UserProfile(
        user=user,
        avatar_url=avatar_url,
        recent_activity=recent_activity,
    )

Retry Pattern

import time
from functools import wraps
from typing import Type


def retry(
    max_attempts: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    exceptions: tuple[Type[Exception], ...] = (Exception,),
):
    """Decorator to retry a function on failure."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            current_delay = delay
            
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        log.warning(
                            f"Attempt {attempt + 1} failed: {e}. "
                            f"Retrying in {current_delay}s..."
                        )
                        time.sleep(current_delay)
                        current_delay *= backoff
            
            raise last_exception
        return wrapper
    return decorator


@retry(max_attempts=3, delay=1.0, exceptions=(ConnectionError, TimeoutError))
def fetch_data_from_api(url: str) -> dict:
    """Fetch data with automatic retry."""
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

Logging

Basic Logging Setup

import logging

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

# Get logger for module
logger = logging.getLogger(__name__)

# Use the logger
logger.debug("Detailed information for debugging")
logger.info("General information about program execution")
logger.warning("Something unexpected happened")
logger.error("Error occurred, but program continues")
logger.critical("Serious error, program may not continue")

Structured Logging

import logging
import json
from datetime import datetime


class JSONFormatter(logging.Formatter):
    """Format log records as JSON."""
    
    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
        }
        
        # Add exception info if present
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)
        
        # Add extra fields
        if hasattr(record, "extra"):
            log_data.update(record.extra)
        
        return json.dumps(log_data)


# Configure structured logging
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())

logger = logging.getLogger("app")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

Logging Best Practices

import logging

logger = logging.getLogger(__name__)


def process_order(order_id: int) -> None:
    """Process an order with proper logging."""
    
    # Use extra for structured context
    log = logging.LoggerAdapter(logger, {"order_id": order_id})
    
    log.info("Starting order processing")
    
    try:
        order = fetch_order(order_id)
        log.info("Order fetched", extra={"customer_id": order.customer_id})
        
        # Don't log sensitive data!
        # BAD: log.info(f"Processing order with card {order.card_number}")
        # GOOD: log.info(f"Processing payment for order")
        
        result = charge_payment(order)
        log.info(
            "Payment processed",
            extra={"transaction_id": result.transaction_id},
        )
        
    except PaymentError as e:
        # Log the error with context
        log.error(
            "Payment failed",
            extra={"error_code": e.code, "error_message": str(e)},
            exc_info=True,  # Include stack trace
        )
        raise
    
    except Exception:
        # Log unexpected errors
        log.exception("Unexpected error during order processing")
        raise
    
    log.info("Order processing completed successfully")

Logging Configuration

# logging_config.py
import logging.config

LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "standard": {
            "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
        },
        "detailed": {
            "format": "%(asctime)s - %(name)s - %(levelname)s - %(module)s:%(lineno)d - %(message)s"
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "standard",
            "stream": "ext://sys.stdout",
        },
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "DEBUG",
            "formatter": "detailed",
            "filename": "app.log",
            "maxBytes": 10485760,  # 10MB
            "backupCount": 5,
        },
        "error_file": {
            "class": "logging.FileHandler",
            "level": "ERROR",
            "formatter": "detailed",
            "filename": "errors.log",
        },
    },
    "loggers": {
        "": {  # Root logger
            "level": "DEBUG",
            "handlers": ["console", "file"],
        },
        "app": {
            "level": "DEBUG",
            "handlers": ["console", "file", "error_file"],
            "propagate": False,
        },
        "urllib3": {
            "level": "WARNING",
        },
    },
}


def setup_logging():
    logging.config.dictConfig(LOGGING_CONFIG)

Context Managers for Logging

import logging
import time
from contextlib import contextmanager

logger = logging.getLogger(__name__)


@contextmanager
def log_operation(operation_name: str, **context):
    """Context manager for logging operation timing."""
    start_time = time.time()
    logger.info(f"Starting {operation_name}", extra=context)
    
    try:
        yield
        duration = time.time() - start_time
        logger.info(
            f"Completed {operation_name}",
            extra={"duration_seconds": duration, **context},
        )
    except Exception as e:
        duration = time.time() - start_time
        logger.error(
            f"Failed {operation_name}",
            extra={"duration_seconds": duration, "error": str(e), **context},
            exc_info=True,
        )
        raise


# Usage
def process_batch(items: list[Item]) -> None:
    with log_operation("batch_processing", item_count=len(items)):
        for item in items:
            with log_operation("item_processing", item_id=item.id):
                process_item(item)

Debugging Strategies

Using the Debugger

# Insert breakpoint in code
def complex_function(data):
    processed = preprocess(data)
    
    breakpoint()  # Python 3.7+ built-in debugger
    
    result = transform(processed)
    return result


# Or use pdb directly
import pdb

def another_function(items):
    for i, item in enumerate(items):
        if item.is_problematic:
            pdb.set_trace()  # Drops into debugger
        process(item)

Debugger Commands

# Common pdb commands
n (next)      - Execute next line
s (step)      - Step into function
c (continue)  - Continue execution
p variable    - Print variable value
pp variable   - Pretty print variable
l (list)      - Show current location in code
w (where)     - Show stack trace
q (quit)      - Quit debugger
h (help)      - Show help

Debug Logging

import logging

logger = logging.getLogger(__name__)


def debug_complex_algorithm(data: list[int]) -> int:
    """Algorithm with debug logging."""
    logger.debug(f"Input data: {data}")
    
    # Step 1
    filtered = [x for x in data if x > 0]
    logger.debug(f"After filtering: {filtered}")
    
    # Step 2
    sorted_data = sorted(filtered, reverse=True)
    logger.debug(f"After sorting: {sorted_data}")
    
    # Step 3
    if not sorted_data:
        logger.debug("Empty data, returning 0")
        return 0
    
    result = sorted_data[0] + sorted_data[-1]
    logger.debug(f"Final result: {result}")
    
    return result

Assertions for Debugging

def calculate_average(numbers: list[float]) -> float:
    """Calculate average with defensive assertions."""
    # Precondition checks
    assert numbers, "numbers list cannot be empty"
    assert all(isinstance(n, (int, float)) for n in numbers), "all elements must be numbers"
    
    total = sum(numbers)
    count = len(numbers)
    
    # Invariant check
    assert count > 0, "count should be positive"
    
    result = total / count
    
    # Postcondition check
    assert min(numbers) <= result <= max(numbers), "average should be within range"
    
    return result


# Assertions can be disabled in production with -O flag
# python -O script.py

Practical Exercise

Exercise 1: Implement Error Handling

# Add proper error handling to this function

def transfer_money(from_account_id: int, to_account_id: int, amount: Decimal) -> None:
    """Transfer money between accounts."""
    from_account = account_repository.get(from_account_id)
    to_account = account_repository.get(to_account_id)
    
    from_account.balance -= amount
    to_account.balance += amount
    
    account_repository.save(from_account)
    account_repository.save(to_account)

# Solution with proper error handling

class InsufficientFundsError(Exception):
    def __init__(self, account_id: int, available: Decimal, requested: Decimal):
        self.account_id = account_id
        self.available = available
        self.requested = requested
        super().__init__(
            f"Insufficient funds in account {account_id}: "
            f"available {available}, requested {requested}"
        )


class AccountNotFoundError(Exception):
    def __init__(self, account_id: int):
        self.account_id = account_id
        super().__init__(f"Account {account_id} not found")


def transfer_money(
    from_account_id: int,
    to_account_id: int,
    amount: Decimal,
) -> TransferResult:
    """Transfer money between accounts.
    
    Raises:
        ValueError: If amount is not positive.
        AccountNotFoundError: If either account doesn't exist.
        InsufficientFundsError: If source account has insufficient funds.
    """
    logger.info(
        "Starting transfer",
        extra={
            "from_account": from_account_id,
            "to_account": to_account_id,
            "amount": str(amount),
        }
    )
    
    # Validate amount
    if amount <= 0:
        raise ValueError("Transfer amount must be positive")
    
    # Get accounts
    from_account = account_repository.get(from_account_id)
    if from_account is None:
        raise AccountNotFoundError(from_account_id)
    
    to_account = account_repository.get(to_account_id)
    if to_account is None:
        raise AccountNotFoundError(to_account_id)
    
    # Check balance
    if from_account.balance < amount:
        raise InsufficientFundsError(
            from_account_id,
            from_account.balance,
            amount,
        )
    
    try:
        # Perform transfer
        from_account.balance -= amount
        to_account.balance += amount
        
        account_repository.save(from_account)
        account_repository.save(to_account)
        
        logger.info(
            "Transfer completed",
            extra={"from_account": from_account_id, "to_account": to_account_id}
        )
        
        return TransferResult(success=True, transaction_id=generate_id())
        
    except DatabaseError as e:
        logger.error(
            "Transfer failed due to database error",
            extra={"error": str(e)},
            exc_info=True,
        )
        raise TransferError("Transfer failed, please try again") from e

Exercise 2: Add Logging

Add comprehensive logging to this service:

class OrderService:
    def create_order(self, customer_id: int, items: list[dict]) -> Order:
        customer = self.customer_repo.get(customer_id)
        order = Order(customer=customer, items=items)
        
        for item in items:
            product = self.product_repo.get(item["product_id"])
            self.inventory_service.reserve(product.id, item["quantity"])
        
        order.total = self.calculate_total(order)
        saved_order = self.order_repo.save(order)
        
        self.notification_service.send_confirmation(customer.email, saved_order)
        
        return saved_order

Error Handling Checklist

• Catch specific exceptions, not broad ones

• Create custom exceptions for domain errors

• Log errors with context

• Don't log sensitive data

• Use structured logging

• Include stack traces for unexpected errors

• Fail fast on invalid input

• Provide graceful degradation where appropriate

Key Takeaways

1. Be specific with exceptions - Catch only what you can handle

2. Create domain exceptions - Custom exceptions make code clearer

3. Log with context - Include relevant data for debugging

4. Never log secrets - Passwords, tokens, and PII stay out of logs

5. Use structured logging - JSON logs are easier to search and analyze

What's Next?

With robust error handling in place, let's automate quality checks. In Article 10: CI/CD Fundamentals, we'll set up GitHub Actions for automated testing, linting, and deployment pipelines.

---

This article is part of the Software Engineering 101 series.