Validation, Contracts & Error Handling

Validation is how a Python data job proves that output rows are trustworthy before they reach a file, API, database, or warehouse.

This chapter turns earlier habits into a full contract system: required fields, field rules, record rules, batch rules, rejected rows, warnings, fatal errors, exception context, manifests, and tests.

The goal is not to make code noisy. The goal is to make failure paths explicit enough that bad data is stopped, explained, and measured.

rejected row

A row excluded from accepted output with structured reason and source context.

warning

A non-blocking quality signal that is reported while the row may continue.

fatal error

A batch-level problem that should stop the run, such as missing required columns.

contract evidence

Counts and reports proving what validation accepted, rejected, warned, or stopped.

A validation rule is a promise about what accepted output means. If amount is required and non-negative, accepted rows must satisfy that promise every time.

Contracts should be close to the boundary they protect: schema validation before reading rows, row validation before normalization, batch validation before writing output.

The contract is incomplete unless it names the failure path.

Field-level validation checks one value at a time: requiredness, type, range, allowed set, nullability, and parseability.

A strong field rule returns a clear reason that names the field and bad value.

Do this before normalization depends on the value, so invalid data cannot become a misleading default.

from decimal import Decimal, InvalidOperation

ALLOWED_STATUSES = {"paid", "refunded", "pending"}

def validate_amount(value: str) -> Decimal:
    if value in {"", None}:
        raise ValueError("amount is required")
    try:
        amount = Decimal(value)
    except InvalidOperation as exc:
        raise ValueError("amount must be decimal") from exc
    if amount < 0:
        raise ValueError("amount must be non-negative")
    return amount

def validate_status(value: str) -> str:
    if value not in ALLOWED_STATUSES:
        raise ValueError("status is not allowed")
    return value

Record-level rules inspect one row: required fields, valid amount, coherent status, or timestamp parseability.

Batch-level rules need more than one row: duplicate keys, minimum row counts, reconciliation totals, or one-current-row-per-entity.

Schema-level rules protect the whole file before row validation begins.

from collections import Counter

order_id_counts = Counter(row["order_id"] for row in raw_orders)
duplicate_keys = [
    {"order_id": order_id, "count": count}
    for order_id, count in order_id_counts.items()
    if count > 1
]

Schema checks make sure the source has the columns or keys required before row logic begins.

For CSV, this often means checking headers. For JSON-like records, it means checking required keys. For typed outputs, it means proving the normalized shape before writing.

A missing required column is usually fatal because every row is affected.

REQUIRED_COLUMNS = {"order_id", "customer_id", "amount", "event_time"}

def validate_headers(headers: set[str]) -> None:
    missing = REQUIRED_COLUMNS - headers
    if missing:
        raise ValueError(f"missing required columns: {sorted(missing)}")

validate_headers({"order_id", "customer_id", "amount"})

Not every issue deserves the same response. A warning reports something unusual while keeping the row. A rejection excludes a bad row. A fatal error stops the batch.

The distinction should be policy, not mood. Unknown status might be fatal in finance, a rejection in order processing, or a warning during a migration.

Once chosen, the policy should be encoded and counted.

outcomes = [
    {"kind": "warning", "field": "coupon_code", "reason": "unknown coupon kept"},
    {"kind": "rejected", "field": "amount", "reason": "amount must be decimal"},
    {"kind": "fatal", "field": "headers", "reason": "event_time column missing"},
]

A rejected-row report is structured evidence for excluded records. It should be easy to join back to the source and easy for a human to understand.

The minimum useful fields are file_name, line_number, field, value, and reason. Add batch_id or source system when needed.

Rejected rows are output, not garbage. They are how the pipeline stays honest.

def reject_row(file_name, line_number, field, value, reason):
    return {
        "file_name": file_name,
        "line_number": line_number,
        "field": field,
        "value": value,
        "reason": reason,
    }

rejected = reject_row("orders.csv", 3, "amount", "bad", "amount must be decimal")

Exceptions are useful for impossible states and fatal boundaries, but row-level validation should often become structured rejected output.

A good strategy is to let small validators raise specific errors, catch them at the row boundary, and convert them into rejected rows with context.

When re-raising, use exception chaining so the low-level cause is not lost.

class ValidationError(Exception):
    pass

def validate_order(row):
    try:
        amount = validate_amount(row["amount"])
    except ValueError as exc:
        raise ValidationError(
            f'line {row["line_number"]} field amount: {exc}'
        ) from exc
    return amount

When a low-level error crosses a boundary, you usually want to re-raise it as something the caller understands — a `ValueError` about a bad amount rather than whatever the parser threw. The question is what happens to the original.

`raise NewError(...) from exc` keeps it, as `__cause__`, and the traceback prints both with "The above exception was the direct cause of the following exception". A bare `raise NewError(...)` inside an `except` block keeps it too, as `__context__` — but nothing marks it as intentional, and the traceback says "During handling of the above exception, another exception occurred", which reads like a bug in your handler.

The third form, a bare `raise` with no argument, re-raises the current exception unchanged with its original traceback. That is the right move when you only wanted to log or clean up and the caller should still see exactly what happened.

__cause__

The exception named by `raise ... from exc`. States that the translation was deliberate.

__context__

Whatever was being handled when this was raised. Set automatically, and reads as accidental.

bare raise

`raise` with no argument, inside `except`: re-raises the current exception with its original traceback.

def parse(raw):
    try:
        return int(raw)
    except ValueError as exc:
        raise ValueError(f"bad amount {raw!r}") from exc   # explicit cause

try:
    parse("12,5")
except ValueError as exc:
    print(type(exc).__name__, exc, "| cause:", type(exc.__cause__).__name__)

def parse_bare(raw):
    try:
        return int(raw)
    except ValueError:
        raise ValueError(f"bad amount {raw!r}")            # no "from"

try:
    parse_bare("12,5")
except ValueError as exc:
    print("cause:", exc.__cause__, "| context:", type(exc.__context__).__name__)

A validation error without context is a puzzle. A validation error with file, line, field, value, and reason is a repair instruction.

Attach context as early as possible. Chapter 4 preserved line numbers during ingestion; this chapter uses them in validation evidence.

Be careful with sensitive values. Sometimes the report should redact raw value while preserving enough detail to debug.

error_context = {
    "file_name": "orders.csv",
    "line_number": 3,
    "field": "amount",
    "value": "bad",
    "reason": "amount must be decimal",
}

A manifest is the run-level proof that validation did what it promised.

It should include rows_seen, accepted, rejected, warning_count, fatal_error, schema_version, and output locations when available.

The manifest should reconcile with the accepted and rejected outputs.

manifest = {
    "rows_seen": 5,
    "accepted": 2,
    "rejected": 2,
    "warning_count": 1,
    "fatal_error": False,
    "schema_version": "orders.v3",
}

After row validation, run final quality checks before writing: duplicate keys, required non-null output columns, row count expectations, and reference coverage.

These checks protect the boundary between validated Python records and durable output.

The correct action depends on the contract. Duplicate primary keys may be fatal; optional warning counts may be reported.

from collections import Counter

accepted_id_counts = Counter(row["order_id"] for row in accepted)
duplicate_output_keys = [
    order_id
    for order_id, count in accepted_id_counts.items()
    if count > 1
]

if duplicate_output_keys:
    raise ValueError(f"duplicate output order_id values: {duplicate_output_keys}")

Validation rules are perfect for small tests because every rule has tiny input and expected output.

Test the happy path, missing required values, bad types, invalid status, duplicate keys, warning-only cases, and fatal schema errors.

Expected rejected rows should match exact field, value, and reason text.

def test_rejects_bad_amount():
    row = {"file_name": "orders.csv", "line_number": 3, "order_id": "1002", "amount": "bad"}

    rejected = validate_row(row)

    assert rejected == {
        "file_name": "orders.csv",
        "line_number": 3,
        "field": "amount",
        "value": "bad",
        "reason": "amount must be decimal",
    }

A strong validation design states schema checks, field checks, record checks, batch checks, failure paths, rejected reports, and manifest evidence.

Use clear language: warnings are kept and reported, rejected rows are excluded and reported, fatal errors stop the run.

This is the intermediate capstone because every later boundary, APIs, databases, concurrency, orchestration, depends on trustworthy validation.

validation_summary = {
    "schema_checked": True,
    "field_rules_checked": ["amount", "event_time", "status", "customer_id"],
    "batch_rules_checked": ["duplicate order_id"],
    "outputs": ["accepted rows", "rejected rows", "warnings", "manifest"],
}

Four exercises for this chapter, in the module workspace. Each one runs against graded cases, and each has a trap that is in the data rather than in the algorithm.

Four exercises where failure is data rather than an exception, and the report is the deliverable.

Do them with the chapter closed. If one goes wrong, come back to the section it belongs to rather than re-reading the whole thing.