Validation, Contracts & Error Handling

In chapter 1 you wrote check_order. It raised ValueError for a bad record, and a loop moved bad records to a rejected list. This chapter builds the full version for a real CSV file.

The file is the shop again, with five bad rows added at the end. Line 1 is the header: the line with the column names. The loader gives each row its line number:

import csv, io

text = """order_id,customer_id,status,amount
101,1,paid,40.0
102,1,paid,25.0
103,2,refunded,60.0
104,3,paid,15.5
105,3,pending,
106,4,paid,80.0
107,5,paid,
108,,shipped,-5.0
109,6,paid,twelve
110,,paid,9.0
111,7,shipped,30.0
"""

reader = csv.DictReader(io.StringIO(text))
rows = [(line_number, row) for line_number, row in enumerate(reader, start=2)]

print(len(rows), "data rows")
print(rows[0])
print(rows[7])

Here are the six habits this chapter teaches:

1. Check each column with its own small function.
2. Collect every problem in a row, not only the first.
3. Sort problems into warnings, rejections and fatal errors.
4. Catch only the errors you expect.
5. Retry only temporary errors.
6. Report and count what you rejected.

A validation rule is a promise about good data. A table of rules is the contract from chapter 1: what must be true, and what happens if it is not.

These are the rules for the shop file:

The next sections turn each line of this table into code. A reviewer can read the table and predict what the code rejects.

Check the header before any row. If a column is missing, every row fails for the same reason. One clear stop is better than thousands of row errors.

We make our own kind of error for this. The code defines a class, which is a new kind of object.

(Exception) says: this is a kind of error. pass means: nothing more to add. The name BatchError means the whole batch is broken:

class BatchError(Exception):
    pass

REQUIRED = ["order_id", "customer_id", "status", "amount"]

def check_header(fieldnames):
    missing = [name for name in REQUIRED if name not in (fieldnames or [])]
    if missing:
        raise BatchError(f"missing columns: {missing}")

short_file = """order_id,customer_id,status
101,1,paid
"""
reader = csv.DictReader(io.StringIO(short_file))
try:
    check_header(reader.fieldnames)
except BatchError as error:
    print("stop:", error)

check_header(csv.DictReader(io.StringIO(text)).fieldnames)
print("header ok")

An empty file has no header at all, and fieldnames is None. The text (fieldnames or []) turns None into an empty list, so the check reports every column as missing.

An extra column is usually harmless. Decide in the table whether it is ignored or gets a warning.

Give each column its own small function. It takes the text from the file. It returns a clean value, or raises ValueError with a short reason.

The file holds only text, so each function also converts the value. Whole numbers become int. Money becomes Decimal, which is exact, unlike a float.

Decimal("twelve") fails with its own error, InvalidOperation. We catch it and raise a ValueError with a clear reason.

In except InvalidOperation as exc, as exc gives the caught error a name. In raise ... from exc, from exc keeps it as the cause. The section "Keep the original cause" explains this.

one_of is a function that returns a function. You give it a set, which is a group of values in { }, and you get back a checker for those values.

The test loop holds pairs of a checker and a text. A function is a value, so it can sit in a list. check.__name__ is the name of the function.

Two checkers are tested here: whole_number and money. one_of is used in the next section. Here is the code, with seven test calls:

from decimal import Decimal, InvalidOperation

def whole_number(raw):
    if raw == "":
        raise ValueError("is required")
    try:
        return int(raw)
    except ValueError as exc:
        raise ValueError("must be a whole number") from exc

def one_of(allowed):
    def check(raw):
        if raw not in allowed:
            raise ValueError("is not allowed")
        return raw
    return check

def money(raw):
    if raw == "":
        return None
    try:
        value = Decimal(raw)
    except InvalidOperation as exc:
        raise ValueError("must be a number") from exc
    if value < 0:
        raise ValueError("must not be negative")
    return value

for check, raw in [(whole_number, "101"), (whole_number, ""), (whole_number, "1O1"),
                   (money, "40.0"), (money, ""), (money, "twelve"), (money, "-5.0")]:
    try:
        print(check.__name__, repr(raw), "->", repr(check(raw)))
    except ValueError as error:
        print(check.__name__, repr(raw), "-> error:", error)

If you stop at the first problem, the source owner fixes one thing, reruns, and finds the next. Collect all problems in one pass instead.

A spec is a dict from column name to checker. The function check_row runs every checker and keeps every problem. It uses whole_number, one_of and money from the last section:

SPEC = {
    "order_id": whole_number,
    "customer_id": whole_number,
    "status": one_of({"paid", "refunded", "pending"}),
    "amount": money,
}

def check_row(line_number, row):
    clean = {}
    problems = []
    for field, check in SPEC.items():
        try:
            clean[field] = check(row[field])
        except ValueError as error:
            problems.append({"line": line_number, "field": field,
                             "value": row[field], "reason": str(error)})
    return clean, problems

for line_number, row in [rows[0], rows[7], rows[8]]:
    clean, problems = check_row(line_number, row)
    print(line_number, clean)
    for problem in problems:
        print("  ", problem)

The try block holds one checker call, and it catches only ValueError. A missing column would raise KeyError, which is not caught on purpose: the header check should have stopped the run already.

A row with too few cells gives None, and int(None) raises TypeError, which is not caught here. Catch short rows when you read the file, as in the files chapter.

A field rule looks at one value. A row rule looks at several values in one row. A batch rule looks at many rows at once.

Our table has a row rule: a paid order must have an amount. Each field is fine alone, but the pair is wrong. This function checks it after check_row passes:

def check_record(line_number, clean):
    if clean["status"] == "paid" and clean["amount"] is None:
        return [{"line": line_number, "field": "amount", "value": "",
                 "reason": "is required when paid"}]
    return []

for line_number, row in rows[3:7]:
    clean, problems = check_row(line_number, row)
    if not problems:
        problems = check_record(line_number, clean)
    result = "ok" if not problems else problems[0]["reason"]
    print(line_number, row["status"], repr(row["amount"]), "->", result)

A batch rule needs all the rows. One row cannot know that its order_id appears twice.

Count the ids with Counter, the tool from the keyed-data chapter. A duplicate is fatal, so the check raises BatchError:

from collections import Counter

def check_unique(ids):
    counts = Counter(ids)
    duplicates = [order_id for order_id, n in counts.items() if n > 1]
    if duplicates:
        raise BatchError(f"duplicate order_id: {duplicates}")

check_unique([101, 102, 103])
print("no duplicates")

try:
    check_unique([101, 102, 103, 102])
except BatchError as error:
    print("stop:", error)

A problem can have three answers. Here they are, with an example from the shop file:

You met three choices in chapter 1: skip, quarantine, fail loud. A rejection is quarantine. A fatal error is fail loud.

A warning is new: you keep the row. We do not use skip, because it hides bad rows.

The answer is a rule you choose, not a guess. An unknown status could be fatal in a finance job and only a warning in a test run. Choose one answer per rule, write it in code, and count each answer.

The function run below applies the policy to every row. It uses check_row and check_record from the earlier sections. A new rule adds a warning for amounts above 70:

WARN_ABOVE = Decimal("70")

def run(rows):
    accepted, warnings, rejected = [], [], []
    for line_number, row in rows:
        clean, problems = check_row(line_number, row)
        if not problems:
            problems = check_record(line_number, clean)
        if problems:
            rejected.extend(problems)
            continue
        if clean["amount"] is not None and clean["amount"] > WARN_ABOVE:
            warnings.append({"line": line_number, "field": "amount",
                             "value": row["amount"], "reason": "is above 70"})
        accepted.append(clean)
    return accepted, warnings, rejected

check_header(csv.DictReader(io.StringIO(text)).fieldnames)
accepted, warnings, rejected = run(rows)
print("accepted", len(accepted))
print("warnings", warnings)
print("rejected problems", len(rejected), "on", len({p["line"] for p in rejected}), "rows")

Nothing in this chapter catches BatchError. If check_header fails, the error stops the job before run starts. A fatal error must stop the job, not become one more rejected row.

An error that says only "invalid row" is hard to use. A good error says where, what and why: the file, the line, the field, the value and the reason.

Add the context when the problem is created, not later. Our problems already hold the line, field, value and reason.

In the f-string below, {x!r} shows the value x with quotes, so an empty value looks like two quotes. The loop also adds the file name:

for problem in rejected[:3]:
    print(f"orders.csv line {problem['line']}, {problem['field']} "
          f"{problem['reason']} (value {problem['value']!r})")

The reason is for the person who fixes the source file. Write it so they can fix the row without reading your code.

For private data, such as an email address, hide the value in the report. Keep the line number and the field.

The try block runs its steps. If an error of the named kind happens, the except block runs instead. The wider the named kind, the more it hides.

A bare except names no kind, so it catches every error, even a typo. except Exception is almost as wide.

A bare except also catches KeyboardInterrupt, the signal that Ctrl+C sends, so Ctrl+C may not stop the job. In this loop, amout is a typo:

def load_bad(rows):
    kept = []
    for line_number, row in rows:
        try:
            kept.append({"order_id": int(row["order_id"]),
                         "amount": Decimal(row["amout"])})
        except:
            pass
    return kept

print("kept", len(load_bad(rows)))

Now catch only the error you expect from a bad amount, which is InvalidOperation. A wrong column name is a different error, and it must be loud:

def load_ok(rows):
    kept = []
    for line_number, row in rows:
        try:
            kept.append({"order_id": int(row["order_id"]),
                         "amount": Decimal(row["amout"])})
        except InvalidOperation:
            pass
    return kept

try:
    load_ok(rows)
except KeyError as error:
    print("KeyError:", error)

Optional: a try block that grew too long

Every step inside a long try block is blamed for the error. Here the rate table has no pending status, so tax_wide gives the wrong reason:

RATES = {"refunded": Decimal("0"), "paid": Decimal("0.20")}

def tax_wide(row):
    try:
        amount = Decimal(row["amount"])
        return amount * RATES[row["status"]]
    except Exception:
        return "reject: bad amount"

def tax_narrow(row):
    try:
        amount = Decimal(row["amount"])
    except InvalidOperation:
        return "reject: bad amount"
    return amount * RATES[row["status"]]

paid_row = rows[3][1]
pending = {"status": "pending", "amount": "5.0"}

print("wide:", tax_wide(paid_row), "|", tax_wide(pending))
print("narrow:", tax_narrow(paid_row))
try:
    tax_narrow(pending)
except KeyError as error:
    print("narrow: KeyError", error)

Optional: you can skip this section on a first read.

Sometimes you catch a low-level error and raise a clearer one. Write raise NewError(...) from exc. It keeps the first error as the cause.

A traceback is the list of lines Python prints when a program stops with an error. Here the first error is InvalidOperation from Decimal, and the caller sees a ValueError in its own words:

def parse(raw):
    try:
        return Decimal(raw)
    except InvalidOperation as exc:
        raise ValueError(f"bad amount {raw!r}") from exc

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

Now leave out the word from. Python still links the two errors, but under a different name:

def parse_plain(raw):
    try:
        return Decimal(raw)
    except InvalidOperation:
        raise ValueError(f"bad amount {raw!r}")

try:
    parse_plain("twelve")
except ValueError as exc:
    print("cause:", exc.__cause__, "| context:", type(exc.__context__).__name__)

Pick the form by what you want the caller to see:

Some errors go away if you try again. A server may be busy for one second. This is a temporary error, also called transient.

Other errors never go away. A file you may not read stays unreadable. A retry of this permanent error only repeats the failure.

So the rule is: retry temporary errors only. This table sorts four errors you will meet:

Wait longer after each failure. This is called backoff, and it gives a busy server time to recover.

A common plan doubles the wait each time, up to a maximum. Here 2 ** n means 2 to the power n. These lines build that schedule:

delays = [min(30, 2 ** n) for n in range(6)]
print(delays)

The function with_retries uses a schedule like this. It retries only TimeoutError and ConnectionError. The last try sits outside the loop, so its error goes up to the caller.

The demo passes sleep as a parameter and saves the waits in a list, so it runs fast:

calls = {"n": 0}

def flaky_fetch():
    calls["n"] += 1
    if calls["n"] < 3:
        raise TimeoutError("server did not answer")
    return "11 rows"

def with_retries(action, delays, sleep):
    for delay in delays:
        try:
            return action()
        except (TimeoutError, ConnectionError) as error:
            print(f"try failed: {error}. wait {delay}s")
            sleep(delay)
    return action()

waited = []
print(with_retries(flaky_fetch, [1, 2, 4], waited.append))
print("calls", calls["n"], "waited", waited)

def no_access():
    raise PermissionError("no permission to read the file")

try:
    with_retries(no_access, [1, 2, 4], waited.append)
except PermissionError as error:
    print("stopped:", error)

def always_down():
    raise TimeoutError("server did not answer")

try:
    with_retries(always_down, [1, 2, 4], waited.append)
except TimeoutError:
    print("gave up")

Optional: you can skip this section on a first read, but exercise 3 uses it.

A list of 7 problems is easy to read. A real file can have a million. Group the problems by field and reason, and put the biggest group first.

The function summarise counts each group with Counter. It keeps only 2 line numbers per group.

sorted(..., key=sort_key) calls sort_key on each item and sorts by what it returns. The key is a tuple: Python compares the first value, and uses the next only on a tie. The minus puts big counts first, and the names break ties, so the report is the same on every run:

def sort_key(item):
    group, count = item
    return (-count, group)

def summarise(problems, sample_size=2):
    counts = Counter((p["field"], p["reason"]) for p in problems)
    ordered = sorted(counts.items(), key=sort_key)
    report = []
    for (field, reason), count in ordered:
        lines = [p["line"] for p in problems
                 if (p["field"], p["reason"]) == (field, reason)]
        report.append({"field": field, "reason": reason,
                       "count": count, "lines": lines[:sample_size]})
    return report

for item in summarise(rejected):
    print(item)

Now the source owner can act. Two rows lack a customer_id, and two have a status that is not allowed. The sample lines say where to look first.

A manifest is a small summary of one run. It says how many rows came in, how many were kept, and how many were rejected or warned.

The counts must add up. The word assert checks a fact. If the fact is false, the program stops with an error.

This code builds the manifest and asserts that rows seen equal accepted plus rejected rows:

manifest = {
    "rows_seen": len(rows),
    "accepted": len(accepted),
    "rejected_rows": len({p["line"] for p in rejected}),
    "warnings": len(warnings),
}
print(manifest)

assert manifest["rows_seen"] == manifest["accepted"] + manifest["rejected_rows"]
print("counts add up")

Save the manifest next to the output. A sudden jump in rejected rows is often the first sign that the source changed its format.

Optional: you can skip this section on a first read.

Before you write the output, ask if the whole run looks right. Zero accepted rows is wrong. So is a very high share of rejected rows.

Both mean the source probably broke. The limit of 20% in the code is only an example. Choose the limit for each job.

In the f-string, :.0% shows a share as a percent, so 0.4545 becomes 45%. This function raises BatchError in those cases:

def check_before_write(rows_seen, accepted, rejected_rows, max_share=0.2):
    if not accepted:
        raise BatchError("no rows were accepted")
    share = rejected_rows / rows_seen
    if share > max_share:
        raise BatchError(
            f"{rejected_rows} of {rows_seen} rows rejected "
            f"({share:.0%}), limit {max_share:.0%}"
        )

try:
    check_before_write(manifest["rows_seen"], accepted, manifest["rejected_rows"])
except BatchError as error:
    print("stop:", error)

check_before_write(100, accepted, 3)
print("ok")

Run this check after the row loop and before the write. Stopping early is cheap. Writing bad output over good data is not.

Optional: you can skip this section on a first read.

A test is a small function that checks one rule. The tool pytest runs every function whose name starts with test. Each test uses assert, as in the manifest section.

Save the tests in a file named test_rules.py, after the code from this chapter. Run them with the command pytest -q.

Test one good case and one bad case for each rule. Check the exact reason and field, not only that something failed. The line with pytest.raises passes only if the code inside it raises that error, with a message that matches the text:

import pytest

def test_money_reads_a_decimal():
    assert money("40.0") == Decimal("40.0")

def test_money_rejects_text():
    with pytest.raises(ValueError, match="must be a number"):
        money("twelve")

def test_row_collects_every_problem():
    row = {"order_id": "108", "customer_id": "", "status": "shipped", "amount": "-5.0"}
    clean, problems = check_row(9, row)
    assert [p["field"] for p in problems] == ["customer_id", "status", "amount"]

Each exercise opens in the Python workspace. You write a function, run it on sample rows, and check the result.

The first three use the habits of this chapter. The fourth is harder: it sorts failures and decides what is safe to retry. The retry section is a good first step for it.

Some sections were optional. If you skipped one, read it before the exercise that needs it: "Summarise the rejections" before exercise 3.