Chapter 8: Error Handling — Writing Robust Business Applications

"The question is not whether your code will encounter unexpected data. It will. The question is whether it will fall apart gracefully or crash spectacularly." — Marcus Webb, Acme Corp IT Department

The Monday Morning Problem

It is 7:03 AM on a Monday. Sandra Chen, VP of Sales at Acme Corp, pours her first coffee and opens her laptop. She is expecting the weekly regional sales summary that Priya Okonkwo's script generates automatically every Monday at 6:45 AM. Instead, she finds an email from the automated system:

Subject: [ERROR] Weekly Sales Report — FAILED

Process exited with error code 1.
Traceback (most recent call last):
  File "weekly_report.py", line 47, in <module>
    revenue = float(row['revenue'])
ValueError: could not convert string to float: 'N/A'

No report. The entire script crashed because one cell in one row of one CSV file contained "N/A" instead of a number. Priya will spend the first hour of her Monday manually pulling the data that her script was supposed to automate.

This scenario plays out in businesses every day. Scripts that work perfectly in testing break the moment they encounter real-world data — missing files, unexpected formats, null values, encoding errors, network timeouts. Every one of these is predictable. Every one of these can be handled gracefully.

This chapter teaches you how to write Python code that does not break when reality fails to cooperate.

8.1 Understanding Exceptions: Python's Error System

When Python encounters a problem it cannot resolve, it raises an exception — a signal that something went wrong. If your code does not catch that signal, Python prints a traceback and stops executing.

You have almost certainly seen this already:

# Attempting to divide by zero
monthly_quota = 0
performance_ratio = 12500 / monthly_quota

ZeroDivisionError: division by zero

# Trying to access a dictionary key that doesn't exist
sales_data = {"Q1": 45000, "Q2": 52000}
q3_sales = sales_data["Q3"]

KeyError: 'Q3'

# Converting a non-numeric string to a number
revenue = float("N/A")

ValueError: could not convert string to float: 'N/A'

Each of these is a different exception type. Understanding the types matters because different exceptions call for different responses.

The Exception Hierarchy

Python's exceptions are organized into a hierarchy — a family tree of error types. At the top sits BaseException, and below it, the exceptions you will encounter day to day all inherit from Exception.

BaseException
├── SystemExit              # Raised by sys.exit()
├── KeyboardInterrupt       # Raised when user presses Ctrl+C
└── Exception               # The parent of almost all errors you'll handle
    ├── ArithmeticError
    │   ├── ZeroDivisionError   # 5 / 0
    │   └── OverflowError       # Number too large for float
    ├── LookupError
    │   ├── KeyError            # dict["missing_key"]
    │   └── IndexError          # list[999] when list has 3 items
    ├── ValueError              # Right type, wrong value: float("N/A")
    ├── TypeError               # Wrong type entirely: "5" + 3
    ├── AttributeError          # Object doesn't have that attribute
    ├── NameError               # Variable doesn't exist
    ├── OSError (IOError)       # File system problems
    │   ├── FileNotFoundError   # File doesn't exist
    │   └── PermissionError     # No access to file
    └── RuntimeError            # General runtime errors

This hierarchy is not merely academic. When you catch a LookupError, you catch both KeyError and IndexError. When you catch Exception, you catch almost everything. Understanding the tree lets you be as specific or as broad as the situation requires.

Why Exception Types Matter in Business

Consider these two real-world problems that look similar on the surface:

• A sales report file cannot be found because the overnight ETL job failed
• A sales report file exists but contains corrupted data

Both will crash your script. But the appropriate response is different. In the first case, you might want to alert the IT team that a data pipeline is broken. In the second, you might want to log which rows failed and continue processing the valid ones. Catching the specific exception type lets you respond appropriately to each situation.

8.2 The try/except Block: Catching Exceptions

The fundamental tool for handling exceptions is the try/except block. The syntax is straightforward:

try:
    # Code that might raise an exception
    result = risky_operation()
except ExceptionType:
    # Code that runs if that exception occurs
    handle_the_problem()

Here is a concrete example from Priya's world:

# Without error handling — crashes on bad data
revenue_str = row["revenue"]
revenue = float(revenue_str)  # Crashes if revenue_str is "N/A"
total_revenue += revenue

# With error handling — logs and continues
try:
    revenue_str = row["revenue"]
    revenue = float(revenue_str)
    total_revenue += revenue
except ValueError:
    print(f"Warning: Could not parse revenue value '{revenue_str}' — skipping row")

When the float() call raises a ValueError, Python immediately jumps to the except block. The rest of the try block (the total_revenue += revenue line) does not execute. After the except block finishes, Python continues with whatever code comes after the entire try/except structure.

A Simple but Complete Example

def get_quarterly_revenue(sales_data: dict, quarter: str) -> float:
    """
    Retrieve revenue for a given quarter from a sales dictionary.
    Returns 0.0 if the quarter is not found.
    """
    try:
        return float(sales_data[quarter])
    except KeyError:
        print(f"Quarter '{quarter}' not found in sales data.")
        return 0.0
    except ValueError:
        print(f"Revenue for '{quarter}' is not a valid number: {sales_data[quarter]}")
        return 0.0


acme_sales = {"Q1": "45000", "Q2": "N/A", "Q3": "61200"}

print(get_quarterly_revenue(acme_sales, "Q1"))   # 45000.0
print(get_quarterly_revenue(acme_sales, "Q2"))   # Logs warning, returns 0.0
print(get_quarterly_revenue(acme_sales, "Q4"))   # Logs warning, returns 0.0

8.3 The Bare except — And Why to Avoid It

It is tempting to write a catch-all handler:

# DO NOT do this
try:
    process_invoice(invoice_data)
except:
    print("Something went wrong.")

This is called a bare except, and it is almost always a mistake. Here is why:

1. It hides bugs. If your code has a typo that causes a NameError, the bare except will silently swallow it. You will get "Something went wrong" when the real problem is a bug in your code.

2. It catches things you never intended to catch. A bare except catches KeyboardInterrupt (Ctrl+C) and SystemExit. This means your script becomes impossible to stop with Ctrl+C — you have to kill the entire terminal.

3. It makes debugging a nightmare. Without knowing what exception occurred, you have no idea where to look.

The correct minimum is except Exception:, which catches all runtime errors but leaves KeyboardInterrupt and SystemExit alone:

# Acceptable as a last resort — but specific is always better
try:
    process_invoice(invoice_data)
except Exception as e:
    print(f"Unexpected error processing invoice: {e}")

But even except Exception should usually be replaced with specific exception types.

8.4 Catching Specific Exceptions

Catching specific exceptions makes your code clearer, more correct, and easier to debug. Python lets you write multiple except clauses, one for each exception type:

import csv


def load_sales_file(filepath: str) -> list[dict]:
    """Load a CSV sales file and return a list of row dictionaries."""
    rows = []
    try:
        with open(filepath, encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for row in reader:
                rows.append(row)
    except FileNotFoundError:
        print(f"Error: The file '{filepath}' does not exist.")
        print("Check that the data export ran successfully before this script.")
    except PermissionError:
        print(f"Error: Permission denied reading '{filepath}'.")
        print("Check that no other program has the file open.")
    except UnicodeDecodeError:
        print(f"Error: '{filepath}' contains characters that cannot be decoded as UTF-8.")
        print("Try re-saving the file with UTF-8 encoding.")
    except Exception as e:
        print(f"Unexpected error loading '{filepath}': {type(e).__name__}: {e}")
    return rows

Notice that each exception type gets its own specific message — one that tells the user not just what went wrong but why and what to do about it. This is the difference between a script that anyone can troubleshoot and one that requires a developer to diagnose.

Catching Multiple Exceptions in One Clause

Sometimes two different exceptions should be handled the same way. You can group them using a tuple:

try:
    value = float(row["amount"]) / float(row["units"])
except (ValueError, ZeroDivisionError):
    print(f"Invalid calculation data in row: {row}")
    value = 0.0

8.5 The else Clause: When Everything Goes Right

The try/except structure has an optional else clause that runs only if no exception was raised in the try block. This is useful when you have code that should run on success but does not need exception protection itself:

def process_payment(amount_str: str, account_id: str) -> bool:
    """
    Attempt to process a payment. Returns True on success, False on failure.
    """
    try:
        amount = float(amount_str)
        if amount <= 0:
            raise ValueError(f"Payment amount must be positive, got {amount}")
    except ValueError as e:
        print(f"Invalid payment amount for account {account_id}: {e}")
        return False
    else:
        # This block only runs if no exception occurred in the try block
        print(f"Payment of ${amount:.2f} queued for account {account_id}")
        record_payment_audit(account_id, amount)
        return True

The else clause makes the intent explicit: "this code only makes sense if the risky part succeeded." It also keeps the except block focused on error handling rather than mixing success and failure paths.

8.6 The finally Clause: Cleanup That Always Runs

The finally clause runs regardless of what happens — whether the try block succeeded, whether an exception was caught, or even whether an uncaught exception is propagating upward. Its purpose is cleanup: releasing resources, closing files, resetting state.

import csv

def export_report(data: list[dict], output_path: str) -> None:
    """Write processed sales data to a CSV file."""
    output_file = None
    try:
        output_file = open(output_path, "w", newline="", encoding="utf-8")
        writer = csv.DictWriter(output_file, fieldnames=data[0].keys())
        writer.writeheader()
        writer.writerows(data)
        print(f"Report written to {output_path}")
    except IndexError:
        print("Error: No data to write — the data list is empty.")
    except PermissionError:
        print(f"Error: Cannot write to '{output_path}' — file may be open in another program.")
    except OSError as e:
        print(f"File system error writing report: {e}")
    finally:
        # This runs no matter what — file always gets closed if it was opened
        if output_file is not None:
            output_file.close()

In modern Python, the with statement handles many cleanup scenarios automatically — the context manager closes the file even if an exception occurs. But finally remains essential for non-context-managed resources: database connections, network sockets, temporary files, locked resources.

import sqlite3

def get_customer_tier(customer_id: str, db_path: str) -> str | None:
    """Retrieve a customer's tier from the database."""
    connection = None
    try:
        connection = sqlite3.connect(db_path)
        cursor = connection.cursor()
        cursor.execute(
            "SELECT tier FROM customers WHERE customer_id = ?",
            (customer_id,)
        )
        result = cursor.fetchone()
        return result[0] if result else None
    except sqlite3.OperationalError as e:
        print(f"Database error retrieving customer {customer_id}: {e}")
        return None
    finally:
        # Always close the connection, even if an exception occurred
        if connection is not None:
            connection.close()

The Full Structure

A try statement can have all four clauses together:

try:
    # Code that might raise an exception
    result = risky_operation()
except SpecificError as e:
    # Runs if SpecificError (or a subclass) was raised
    handle_error(e)
except AnotherError:
    # Runs if AnotherError was raised (and not caught above)
    handle_differently()
else:
    # Runs only if NO exception was raised
    use_result(result)
finally:
    # Runs no matter what
    cleanup()

8.7 Raising Exceptions: Signaling Problems

Sometimes you need to raise an exception yourself — when your code detects a problem that Python's built-in checks would not catch.

def calculate_commission(revenue: float, rate: float) -> float:
    """
    Calculate sales commission.

    Args:
        revenue: Total sales revenue (must be non-negative)
        rate: Commission rate as a decimal (must be between 0 and 1)

    Returns:
        Commission amount

    Raises:
        ValueError: If revenue is negative or rate is outside 0-1 range
    """
    if revenue < 0:
        raise ValueError(f"Revenue cannot be negative, got {revenue}")
    if not 0 <= rate <= 1:
        raise ValueError(f"Commission rate must be between 0 and 1, got {rate}")
    return revenue * rate

The raise statement takes an exception instance. When someone calls this function with invalid data, they get a clear error message:

commission = calculate_commission(50000, 0.08)   # Works: 4000.0
commission = calculate_commission(-5000, 0.08)   # Raises: ValueError: Revenue cannot be negative, got -5000
commission = calculate_commission(50000, 8.0)    # Raises: ValueError: Commission rate must be between 0 and 1, got 8.0

Re-raising Exceptions

Sometimes you want to catch an exception, do something (like log it), and then let it continue propagating:

import logging

def load_critical_config(filepath: str) -> dict:
    """Load a configuration file that the application cannot run without."""
    try:
        with open(filepath) as f:
            import json
            return json.load(f)
    except FileNotFoundError:
        logging.critical(
            f"Cannot start application: configuration file '{filepath}' not found. "
            "Contact IT to restore this file."
        )
        raise  # Re-raise the same exception after logging it

The bare raise (with no arguments) re-raises the exception that was just caught, preserving the original traceback. This is how you can log an error and still let the caller know something went wrong.

8.8 Custom Exceptions: Building a Business Error Vocabulary

Python's built-in exceptions cover general programming errors. For business logic errors, you should create your own exception classes. This gives your errors meaningful names and makes your code self-documenting.

Custom exceptions are created by inheriting from Exception (or a more specific built-in exception):

class BusinessError(Exception):
    """Base class for all Acme Corp business logic errors."""
    pass


class ValidationError(BusinessError):
    """Raised when business data fails validation."""

    def __init__(self, field_name: str, value: object, message: str) -> None:
        self.field_name = field_name
        self.value = value
        self.message = message
        super().__init__(f"Validation failed for '{field_name}': {message} (got: {value!r})")


class InvoiceProcessingError(BusinessError):
    """Raised when an invoice cannot be processed."""

    def __init__(self, invoice_id: str, reason: str) -> None:
        self.invoice_id = invoice_id
        self.reason = reason
        super().__init__(f"Cannot process invoice {invoice_id}: {reason}")


class RegionNotFoundError(BusinessError):
    """Raised when a sales region does not exist in the system."""

    def __init__(self, region_code: str) -> None:
        self.region_code = region_code
        super().__init__(
            f"Region '{region_code}' is not a recognized sales region. "
            f"Valid regions are: WEST, EAST, CENTRAL, SOUTH"
        )

With custom exceptions, your code reads like a business document:

def process_invoice(invoice_id: str, amount: float, region: str) -> dict:
    valid_regions = {"WEST", "EAST", "CENTRAL", "SOUTH"}

    if amount <= 0:
        raise ValidationError("amount", amount, "Invoice amount must be positive")

    if region not in valid_regions:
        raise RegionNotFoundError(region)

    # ... processing logic ...
    return {"invoice_id": invoice_id, "amount": amount, "status": "processed"}


# The calling code can catch exactly what it expects:
try:
    result = process_invoice("INV-2024-0847", -500, "WEST")
except ValidationError as e:
    print(f"Data problem: {e}")
    print(f"Field: {e.field_name}, Value: {e.value}")
except RegionNotFoundError as e:
    print(f"Region problem: {e}")
    print(f"Unknown region: {e.region_code}")
except BusinessError as e:
    # Catch any other business errors we didn't anticipate
    print(f"Business logic error: {e}")

Why Custom Exceptions Are Worth the Effort

Consider this comparison:

Without custom exceptions:

except ValueError:
    print("Something is wrong with the invoice data")

With custom exceptions:

except ValidationError as e:
    send_alert_to_data_team(field=e.field_name, bad_value=e.value, message=e.message)
    log_to_audit_trail(invoice_id=invoice_id, error=str(e))

The second version is actionable. The first is a dead end.

8.9 The logging Module: Replacing print() in Production Code

Throughout this chapter you have seen print() calls for error messages. In scripts you run on your own laptop, print() is fine. In production systems — scripts that run on servers, scheduled jobs, shared automation — you need the logging module.

The logging module provides: - Log levels — messages are categorized by severity - Timestamps — every message is automatically timestamped - Log files — messages can be written to a file, not just the screen - Structured output — consistent format across your entire application - Filtering — in production, you can suppress DEBUG messages without changing code

Log Levels

Python's logging module defines five standard severity levels:

Basic Logging Setup

import logging

# Configure logging — do this once, near the top of your script
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s | %(levelname)-8s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

logger = logging.getLogger(__name__)

# Using the logger
logger.debug("Starting invoice processing loop")
logger.info("Processing invoice INV-2024-0847")
logger.warning("Invoice INV-2024-0848 has no tax ID — using default rate")
logger.error("Failed to process invoice INV-2024-0849: amount field is empty")
logger.critical("Database connection lost — cannot continue processing")

Output:

2024-11-18 09:23:41 | DEBUG    | Starting invoice processing loop
2024-11-18 09:23:41 | INFO     | Processing invoice INV-2024-0847
2024-11-18 09:23:41 | WARNING  | Invoice INV-2024-0848 has no tax ID — using default rate
2024-11-18 09:23:41 | ERROR    | Failed to process invoice INV-2024-0849: amount field is empty
2024-11-18 09:23:41 | CRITICAL | Database connection lost — cannot continue processing

Logging to Both Console and File

For scheduled scripts, you almost always want logs written to a file so you can review them later:

import logging
from pathlib import Path


def setup_logging(log_file: str = "sales_report.log") -> logging.Logger:
    """
    Configure logging to write to both the console and a log file.
    Returns a configured logger.
    """
    logger = logging.getLogger("acme_sales")
    logger.setLevel(logging.DEBUG)

    # Formatter — the same format for both handlers
    formatter = logging.Formatter(
        fmt="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    # Console handler — show INFO and above on screen
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(formatter)

    # File handler — capture everything including DEBUG
    file_handler = logging.FileHandler(log_file, encoding="utf-8")
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(formatter)

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

    return logger


logger = setup_logging("monday_report.log")
logger.info("Weekly sales report process started")

Logging Exceptions with Tracebacks

When you catch an exception and want to log the full traceback (not just the message), use logger.exception() or pass exc_info=True:

try:
    result = parse_regional_data(filepath)
except Exception as e:
    # logger.exception() automatically includes the traceback
    logger.exception(f"Unexpected error parsing '{filepath}'")
    # This is equivalent to:
    # logger.error(f"Unexpected error parsing '{filepath}'", exc_info=True)

The traceback in the log file is invaluable when debugging a script that ran overnight and you cannot reproduce the error interactively.

8.10 Defensive Programming: Validating Business Data

"Defensive programming" means writing code that anticipates bad input rather than assuming everything will be correct. This philosophy is especially important in business applications, where data comes from spreadsheets maintained by humans, APIs that occasionally return unexpected formats, and legacy systems with inconsistent standards.

Validate Before You Process

Rather than letting a bad value crash your script halfway through a loop, validate your data before entering the processing loop:

import logging
from pathlib import Path

logger = logging.getLogger(__name__)

REQUIRED_COLUMNS = {"invoice_id", "customer_id", "amount", "date", "region"}


def validate_csv_structure(filepath: str) -> bool:
    """
    Check that a CSV file has the expected columns before processing.
    Returns True if valid, False if not.
    """
    import csv

    try:
        with open(filepath, encoding="utf-8") as f:
            reader = csv.DictReader(f)
            actual_columns = set(reader.fieldnames or [])
    except FileNotFoundError:
        logger.error(f"File not found: {filepath}")
        return False
    except Exception as e:
        logger.error(f"Cannot read file '{filepath}': {e}")
        return False

    missing = REQUIRED_COLUMNS - actual_columns
    if missing:
        logger.error(
            f"File '{filepath}' is missing required columns: {sorted(missing)}. "
            f"Found columns: {sorted(actual_columns)}"
        )
        return False

    logger.info(f"File '{filepath}' structure validated successfully.")
    return True

Validate Individual Values

Create reusable validation functions that raise informative exceptions:

from datetime import datetime


def parse_amount(value: str, row_identifier: str) -> float | None:
    """
    Parse a currency amount from a string.

    Handles: "1234.56", "$1,234.56", "1234", ""
    Returns None for invalid values (caller decides how to handle).
    """
    if not value or value.strip() in ("", "N/A", "None", "null", "-"):
        logger.warning(f"Row {row_identifier}: Amount field is empty or null.")
        return None

    # Remove common currency formatting
    cleaned = value.strip().lstrip("$").replace(",", "")

    try:
        amount = float(cleaned)
    except ValueError:
        logger.warning(
            f"Row {row_identifier}: Cannot parse amount '{value}' as a number. Skipping."
        )
        return None

    if amount < 0:
        logger.warning(
            f"Row {row_identifier}: Negative amount {amount} — "
            "this may be a credit or data error. Flagging for review."
        )

    return amount


def parse_date(value: str, row_identifier: str, fmt: str = "%Y-%m-%d") -> datetime | None:
    """Parse a date string. Returns None for invalid values."""
    if not value or value.strip() in ("", "N/A", "None"):
        logger.warning(f"Row {row_identifier}: Date field is empty.")
        return None

    try:
        return datetime.strptime(value.strip(), fmt)
    except ValueError:
        logger.warning(
            f"Row {row_identifier}: Cannot parse date '{value}' "
            f"using format '{fmt}'. Expected format: {fmt}"
        )
        return None

8.11 Graceful Degradation: Partial Success Is Better Than Total Failure

Priya's report crashed because of one bad row. The other 847 rows were fine, but she got zero output. This is the worst possible failure mode.

Graceful degradation means that when part of your pipeline fails, the rest continues. The output may be incomplete, but it is better than nothing. And critically, the failures are logged so someone can fix the data.

The pattern looks like this:

import logging
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)


@dataclass
class ProcessingResult:
    """Summary of a batch processing run."""
    processed: int = 0
    skipped: int = 0
    errors: list[str] = field(default_factory=list)

    @property
    def total_attempted(self) -> int:
        return self.processed + self.skipped

    def add_error(self, row_id: str, error: str) -> None:
        self.skipped += 1
        self.errors.append(f"{row_id}: {error}")

    def log_summary(self) -> None:
        logger.info(
            f"Processing complete: {self.processed} rows succeeded, "
            f"{self.skipped} rows skipped out of {self.total_attempted} total."
        )
        if self.errors:
            logger.warning(f"Errors encountered ({len(self.errors)}):")
            for error in self.errors:
                logger.warning(f"  - {error}")


def process_invoice_batch(rows: list[dict]) -> ProcessingResult:
    """
    Process a batch of invoice rows.
    Logs and skips invalid rows rather than crashing.
    """
    result = ProcessingResult()
    processed_invoices = []

    for i, row in enumerate(rows, start=1):
        row_id = row.get("invoice_id", f"row_{i}")

        try:
            # Validate and parse each field
            amount_str = row.get("amount", "")
            amount = float(amount_str)

            region = row.get("region", "").strip().upper()
            if region not in {"WEST", "EAST", "CENTRAL", "SOUTH"}:
                raise ValueError(f"Unknown region: '{region}'")

            customer_id = row.get("customer_id", "").strip()
            if not customer_id:
                raise ValueError("customer_id is empty")

            # If we get here, the row is valid
            processed_invoices.append({
                "invoice_id": row_id,
                "amount": amount,
                "region": region,
                "customer_id": customer_id,
            })
            result.processed += 1
            logger.debug(f"Row {row_id}: processed successfully (${amount:.2f})")

        except ValueError as e:
            result.add_error(row_id, str(e))
            logger.warning(f"Row {row_id}: skipped — {e}")
        except KeyError as e:
            result.add_error(row_id, f"Missing field: {e}")
            logger.warning(f"Row {row_id}: skipped — missing field {e}")
        except Exception as e:
            result.add_error(row_id, f"Unexpected error: {type(e).__name__}: {e}")
            logger.error(f"Row {row_id}: unexpected error — {e}", exc_info=True)

    result.log_summary()
    return result

8.12 A Complete Example: The Robust CSV Processor

Let us put everything together. Priya needs a CSV processor that: 1. Checks whether the file exists 2. Validates the column structure 3. Processes each row, logging and skipping bad ones 4. Writes a clean output file 5. Reports a summary of what happened

This is the full solution — the same code you will find in code/robust_csv_processor.py:

"""
robust_csv_processor.py
-----------------------
Production-quality CSV processor for Acme Corp sales data.
Handles missing files, encoding errors, invalid data, and partial failures
without crashing — logs all issues and reports a processing summary.

Usage:
    python robust_csv_processor.py --input sales_data.csv --output clean_sales.csv
"""

import csv
import logging
import argparse
from datetime import datetime
from pathlib import Path
from dataclasses import dataclass, field


# ---------------------------------------------------------------------------
# Logging configuration
# ---------------------------------------------------------------------------

def configure_logging(log_file: str | None = None) -> logging.Logger:
    """Set up logging to console and optionally to a file."""
    logger = logging.getLogger("csv_processor")
    logger.setLevel(logging.DEBUG)

    formatter = logging.Formatter(
        fmt="%(asctime)s | %(levelname)-8s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    console = logging.StreamHandler()
    console.setLevel(logging.INFO)
    console.setFormatter(formatter)
    logger.addHandler(console)

    if log_file:
        fh = logging.FileHandler(log_file, encoding="utf-8")
        fh.setLevel(logging.DEBUG)
        fh.setFormatter(formatter)
        logger.addHandler(fh)

    return logger


logger = configure_logging("csv_processor.log")


# ---------------------------------------------------------------------------
# Custom exceptions
# ---------------------------------------------------------------------------

class CSVProcessorError(Exception):
    """Base exception for CSV processor errors."""


class FileStructureError(CSVProcessorError):
    """Raised when the CSV file has an unexpected structure."""


class RowValidationError(CSVProcessorError):
    """Raised when a row fails validation — row is skipped, not fatal."""


# ---------------------------------------------------------------------------
# Data model
# ---------------------------------------------------------------------------

REQUIRED_COLUMNS = {"invoice_id", "customer_id", "amount", "date", "region"}
VALID_REGIONS = {"WEST", "EAST", "CENTRAL", "SOUTH"}
DATE_FORMAT = "%Y-%m-%d"


@dataclass
class InvoiceRow:
    """A validated, parsed invoice row."""
    invoice_id: str
    customer_id: str
    amount: float
    date: datetime
    region: str


@dataclass
class ProcessingReport:
    """Summary statistics for a processing run."""
    input_file: str
    output_file: str
    rows_processed: int = 0
    rows_skipped: int = 0
    total_revenue: float = 0.0
    errors: list[str] = field(default_factory=list)
    started_at: datetime = field(default_factory=datetime.now)
    finished_at: datetime | None = None

    def record_error(self, row_id: str, message: str) -> None:
        self.rows_skipped += 1
        self.errors.append(f"{row_id}: {message}")

    def finalize(self) -> None:
        self.finished_at = datetime.now()

    def print_summary(self) -> None:
        duration = (
            (self.finished_at - self.started_at).total_seconds()
            if self.finished_at
            else 0
        )
        logger.info("=" * 60)
        logger.info("PROCESSING SUMMARY")
        logger.info("=" * 60)
        logger.info(f"  Input file   : {self.input_file}")
        logger.info(f"  Output file  : {self.output_file}")
        logger.info(f"  Rows OK      : {self.rows_processed}")
        logger.info(f"  Rows skipped : {self.rows_skipped}")
        logger.info(f"  Total revenue: ${self.total_revenue:,.2f}")
        logger.info(f"  Duration     : {duration:.2f}s")
        if self.errors:
            logger.warning(f"  Errors ({len(self.errors)}):")
            for err in self.errors:
                logger.warning(f"    - {err}")
        logger.info("=" * 60)


# ---------------------------------------------------------------------------
# Validation functions
# ---------------------------------------------------------------------------

def validate_row(row: dict[str, str], row_num: int) -> InvoiceRow:
    """
    Validate and parse a raw CSV row into an InvoiceRow.
    Raises RowValidationError if any field is invalid.
    """
    row_id = row.get("invoice_id", f"row_{row_num}").strip()

    # invoice_id
    if not row_id:
        raise RowValidationError("invoice_id is empty")

    # customer_id
    customer_id = row.get("customer_id", "").strip()
    if not customer_id:
        raise RowValidationError(f"{row_id}: customer_id is empty")

    # amount
    amount_raw = row.get("amount", "").strip()
    if not amount_raw or amount_raw.lower() in ("n/a", "null", "none", "-"):
        raise RowValidationError(f"{row_id}: amount is missing or null ('{amount_raw}')")
    try:
        amount = float(amount_raw.lstrip("$").replace(",", ""))
    except ValueError:
        raise RowValidationError(
            f"{row_id}: amount '{amount_raw}' is not a valid number"
        )
    if amount < 0:
        raise RowValidationError(f"{row_id}: amount {amount} is negative")

    # date
    date_raw = row.get("date", "").strip()
    if not date_raw:
        raise RowValidationError(f"{row_id}: date is empty")
    try:
        date = datetime.strptime(date_raw, DATE_FORMAT)
    except ValueError:
        raise RowValidationError(
            f"{row_id}: date '{date_raw}' does not match expected format {DATE_FORMAT}"
        )

    # region
    region = row.get("region", "").strip().upper()
    if region not in VALID_REGIONS:
        raise RowValidationError(
            f"{row_id}: region '{region}' is not valid. "
            f"Expected one of: {sorted(VALID_REGIONS)}"
        )

    return InvoiceRow(
        invoice_id=row_id,
        customer_id=customer_id,
        amount=amount,
        date=date,
        region=region,
    )


# ---------------------------------------------------------------------------
# File loading
# ---------------------------------------------------------------------------

def load_csv(filepath: str) -> tuple[list[dict], bool]:
    """
    Load a CSV file and return (rows, success).
    Handles FileNotFoundError, PermissionError, UnicodeDecodeError.
    """
    path = Path(filepath)

    if not path.exists():
        logger.error(
            f"File not found: '{filepath}'. "
            "Verify the data export completed before running this script."
        )
        return [], False

    if not path.is_file():
        logger.error(f"'{filepath}' is a directory, not a file.")
        return [], False

    # Try UTF-8 first, fall back to latin-1
    for encoding in ("utf-8", "utf-8-sig", "latin-1"):
        try:
            with open(filepath, encoding=encoding, newline="") as f:
                reader = csv.DictReader(f)
                rows = list(reader)
                fieldnames = set(reader.fieldnames or [])

            logger.info(f"Loaded '{filepath}' with encoding '{encoding}' ({len(rows)} rows)")

            # Validate columns
            missing = REQUIRED_COLUMNS - fieldnames
            if missing:
                logger.error(
                    f"File is missing required columns: {sorted(missing)}. "
                    f"Found: {sorted(fieldnames)}"
                )
                return [], False

            return rows, True

        except UnicodeDecodeError:
            logger.debug(f"Encoding '{encoding}' failed for '{filepath}', trying next.")
            continue
        except PermissionError:
            logger.error(
                f"Permission denied: cannot read '{filepath}'. "
                "Close the file in Excel or other applications and try again."
            )
            return [], False
        except csv.Error as e:
            logger.error(f"CSV parsing error in '{filepath}': {e}")
            return [], False

    logger.error(f"Could not decode '{filepath}' with any supported encoding.")
    return [], False


# ---------------------------------------------------------------------------
# Main processing function
# ---------------------------------------------------------------------------

def process_csv(input_path: str, output_path: str) -> ProcessingReport:
    """
    Main entry point: load, validate, and process a sales CSV file.
    """
    report = ProcessingReport(input_file=input_path, output_file=output_path)
    logger.info(f"Starting CSV processing: '{input_path}' -> '{output_path}'")

    # Load the file
    raw_rows, loaded_ok = load_csv(input_path)
    if not loaded_ok:
        logger.critical("Cannot proceed — file could not be loaded.")
        report.finalize()
        report.print_summary()
        return report

    if not raw_rows:
        logger.warning("File loaded successfully but contains no data rows.")
        report.finalize()
        report.print_summary()
        return report

    # Process each row
    valid_rows: list[InvoiceRow] = []
    for row_num, raw_row in enumerate(raw_rows, start=2):  # start=2: row 1 is header
        try:
            invoice = validate_row(raw_row, row_num)
            valid_rows.append(invoice)
            report.rows_processed += 1
            report.total_revenue += invoice.amount
            logger.debug(
                f"Row {row_num} ({invoice.invoice_id}): "
                f"OK — ${invoice.amount:.2f} [{invoice.region}]"
            )
        except RowValidationError as e:
            report.record_error(f"row_{row_num}", str(e))
            logger.warning(f"Row {row_num}: SKIPPED — {e}")

    # Write output
    if valid_rows:
        try:
            with open(output_path, "w", newline="", encoding="utf-8") as out:
                writer = csv.writer(out)
                writer.writerow(["invoice_id", "customer_id", "amount", "date", "region"])
                for inv in valid_rows:
                    writer.writerow([
                        inv.invoice_id,
                        inv.customer_id,
                        f"{inv.amount:.2f}",
                        inv.date.strftime(DATE_FORMAT),
                        inv.region,
                    ])
            logger.info(f"Wrote {len(valid_rows)} clean rows to '{output_path}'")
        except PermissionError:
            logger.error(
                f"Cannot write to '{output_path}' — file may be open in another program."
            )
        except OSError as e:
            logger.error(f"File system error writing output: {e}")
    else:
        logger.warning("No valid rows to write — output file not created.")

    report.finalize()
    report.print_summary()
    return report


# ---------------------------------------------------------------------------
# Command-line interface
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Robust CSV processor for Acme Corp sales data."
    )
    parser.add_argument("--input", required=True, help="Path to input CSV file")
    parser.add_argument("--output", required=True, help="Path to output CSV file")
    args = parser.parse_args()

    result = process_csv(args.input, args.output)

    # Exit with a non-zero code if nothing was processed
    if result.rows_processed == 0:
        raise SystemExit(1)

8.13 Patterns and Anti-Patterns Summary

Anti-Patterns to Avoid

The Bare Except:

# Never do this
try:
    process_data()
except:
    pass

Swallowing Exceptions Silently:

# Worse — the error disappears completely
try:
    amount = float(value)
except ValueError:
    pass

Catching Exception When You Mean a Specific Type:

# Too broad — catches MemoryError, SystemError, and more
try:
    parse_csv(data)
except Exception:
    print("Problem with CSV")

Using Exceptions for Control Flow:

# Exceptions are for exceptional situations, not normal logic
try:
    tier = customer_tiers[customer_id]
except KeyError:
    tier = "standard"  # This is fine if KeyError is genuinely exceptional

# Prefer this when the missing key is a normal, expected outcome:
tier = customer_tiers.get(customer_id, "standard")

Good Patterns to Adopt

Specific Exceptions with Helpful Messages:

try:
    amount = float(row["amount"])
except ValueError:
    logger.warning(
        f"Invoice {row['id']}: amount '{row['amount']}' is not numeric — skipping"
    )
    continue

Validate Before Processing:

# Check everything before the loop, not inside it
if not validate_file_structure(filepath):
    logger.error("File structure invalid — aborting")
    return

for row in rows:
    process_row(row)

Custom Exceptions for Business Logic:

class InvoiceAmountError(ValidationError):
    pass

raise InvoiceAmountError("amount", value, "must be positive")

Log Exceptions with Context:

try:
    result = api_call(customer_id)
except requests.Timeout:
    logger.error(
        f"API timeout for customer {customer_id} after {TIMEOUT}s. "
        "Will retry on next run."
    )

8.14 Chapter Summary

Error handling is not a defensive afterthought — it is the difference between a script you can rely on in production and one that creates emergencies. The key concepts from this chapter:

• Python raises exceptions when it encounters problems it cannot resolve on its own. Understanding the exception hierarchy helps you catch the right things.
• The try/except/else/finally structure gives you full control over what happens when errors occur.
• Always catch specific exception types, never bare except:. Bare except is a bug-hiding machine.
• Use multiple except clauses to respond differently to different problems.
• The else clause runs only on success; finally always runs (use it for cleanup).
• raise lets you signal problems in your own code. Custom exceptions make your error vocabulary match your business domain.
• The logging module replaces print() in production code. Use it from day one.
• Graceful degradation means logging bad rows and continuing, rather than crashing and producing zero output.
• Validate before processing: check file structure, column names, and data types before committing to a processing loop.

In Chapter 9, we will apply these patterns to working with external APIs — where network failures, authentication errors, and unexpected response formats make error handling not just good practice but an absolute requirement.

Next: Chapter 9 — Working with External APIs and Web Data