Capstone Part One: Assembling the Digest

By the end of this lesson you have the shape of the whole project: one entry point, five stages that each take a value and return a value, and a digest whose totals can be checked against the number of records that went in.

Five techniques that each work are not a workflow. The thing that makes them one is the seam between them, and the seam is where projects like this usually come apart.

The failure mode has a recognisable shape: stage two writes into a dictionary that stage four also writes into, stage three reads a global that stage one set, and by the time somebody wants to test stage four alone they cannot, because stage four does not take an input — it takes the accumulated state of everything before it. At that point the only way to test anything is to run all of it, and the only way to find a defect is to read all of it.

The alternative costs nothing at the time and is the difference between a project you can change and one you can only rewrite: every stage takes a value and returns a value. Collection returns files. Parsing returns records and rejections. Validation returns accepted and quarantined. Aggregation returns totals. Reporting returns lines. Each one can be run on its own, with hand-made input, in a test that takes a millisecond.

The digest also needs to be checkable. A report saying "north: 2 orders, 21 units" is a claim, and the only way anybody can believe it is if the same run also says how many records it read, how many it accepted, and how many it set aside — and if those numbers add up.

The layout. A project somebody else can read has an obvious front door:

order-digest/
  run.py              the entry point; parses arguments, calls the pipeline, exits with a status
  digest/
    collect.py        inbox -> list of files
    parse.py          rows -> records, rejections
    validate.py       records -> accepted, quarantined
    aggregate.py      accepted -> totals
    report.py         totals + counts -> report lines
    record.py         everything -> the run record
  settings.toml       paths, budget, deadline, schedule interval
  fixtures/           the recorded inputs the tests run against
  tests/
  RUNBOOK.md

Nothing here is required by Python; it is required by the person reading it in a year. The one rule worth insisting on is that run.py contains no domain transformations: it reads settings, parses arguments, calls stages, and decides the exit status. Keep parsing, validation, and aggregation in functions that can be tested without running the whole program.

If you would rather start from something running, starter_daily_digest.py is this shape collapsed into one file, with four functions left for you and a completed version to compare against afterwards. It reads sample-orders.csv, which is the same data as the capstone's first fixture, so the totals it produces are the ones the capstone's expected output states.

The pipeline, as one function. The five processing stages are collect, parse, validate, aggregate, and report. The run record is a separate final output.

files      = collect(inbox, archive)
records    = parse(rows_from(files), schema)
checked    = validate(records.accepted, rules, lookups)
totals     = aggregate(checked.accepted)
counts     = counts_from(records, checked)
lines      = report(totals, counts, metadata)
record     = finish(job, counts, seconds, status)

This is a map of value handoffs, not executable code or a prescribed function API: rows_from, counts_from, and finish stand for work the entry point must arrange. Keep the transformation stages independent of shared state. Collection may archive input, while writing report lines or a run record belongs at an explicit output boundary.

Counts flow forward, and they are the reconciliation. Each stage reports what it received and what it produced, and the final record adds them up:

read        rows that came out of the files
rejected    rows parse could not turn into records
quarantined records validate set aside
accepted    records that reached the aggregation
read == rejected + quarantined + accepted

That equation is the thing to check in code. A digest that cannot state it is a digest nobody can check.

Settings, in one place. The inbox path, the archive path, the request budget, the deadline, the schedule interval, and the region lookup all belong in one file that is read once, at the start, by the entry point. Values scattered through the stages are values nobody can find when the job needs to move to another machine.

One exit status. 0 for a completed run, and something non-zero for anything else, so a scheduler can tell. A refused run is not a crash, and giving it its own status — 2, say — lets monitoring distinguish "did not run because the input was not there" from "broke".

The middle of the pipeline, joined up:

def parse_rows(rows, regions):
    """Rows to typed records, or a rejection naming the field."""
    accepted, rejected = [], []
    for row in rows:
        try:
            units = int(row["units"])
        except (KeyError, ValueError):
            rejected.append((row.get("order_id", "?"), f"units: {row.get('units', '')!r} is not a whole number"))
            continue
        if row["region_code"] not in regions:
            rejected.append((row["order_id"], f"region_code: {row['region_code']} is not a known region"))
            continue
        accepted.append({**row, "units": units, "group": regions[row["region_code"]]})
    return accepted, rejected


def aggregate(records):
    """Totals by reporting group, from accepted records only."""
    totals = {}
    for record in records:
        group = totals.setdefault(record["group"], {"orders": 0, "units": 0, "value": 0.0})
        group["orders"] += 1
        group["units"] += record["units"]
        group["value"] += record["units"] * float(record["unit_price"])
    return totals


REGIONS = {"NW": "north", "NE": "north", "SW": "south", "SE": "south"}
ROWS = [
    {"order_id": "ORD-1", "region_code": "NW", "units": "12", "unit_price": "4.50"},
    {"order_id": "ORD-2", "region_code": "ZZ", "units": "3", "unit_price": "19.99"},
    {"order_id": "ORD-3", "region_code": "NE", "units": "two", "unit_price": "7.25"},
]

accepted, rejected = parse_rows(ROWS, REGIONS)
totals = aggregate(accepted)
print("read:", len(ROWS))
print("accepted:", len(accepted))
print("rejected:", len(rejected))
for group in sorted(totals):
    figures = totals[group]
    print(f"{group}: {figures['orders']} orders, {figures['units']} units, {figures['value']:.2f}")
for order_id, reason in rejected:
    print("rejected:", order_id, reason)

parse_rows returns two lists for the shown bad-units and unknown-region cases, so those rows do not stop this batch. Returning them lets the caller count both outcomes. This small example does not handle every malformed row: missing region_code or order_id can still raise, and aggregate can raise on an invalid unit_price. A production input contract would need to handle those cases deliberately.

aggregate takes only the accepted records. It never sees a rejection, so it cannot accidentally include one, and it can be tested with three hand-written dictionaries.

{**row, "units": units, ...} builds a new record rather than modifying the row in place. The caller still holds the original, unmodified, which matters when a later stage wants to report what actually arrived.

:.2f formats the displayed money value to two decimal places. Floating-point arithmetic can sometimes produce longer raw decimal representations; this formatting keeps the report readable. For money that must be exact across many operations, use decimal arithmetic rather than relying on display formatting.

Predict every line, in order.

read: 3
accepted: 1
rejected: 2
north: 1 orders, 12 units, 54.00
rejected: ORD-2 region_code: ZZ is not a known region
rejected: ORD-3 units: 'two' is not a whole number

Make aggregate take all the rows instead of only the accepted ones, by calling aggregate(ROWS).

KeyError: 'group'

An assistant was asked to "put the pipeline together into one script". It produced this.

TOTALS = {}
ERRORS = []

def process(rows):
    for row in rows:
        try:
            TOTALS.setdefault(row["region_code"], 0)
            TOTALS[row["region_code"]] += int(row["units"])
        except Exception as error:
            ERRORS.append(str(error))

def main():
    process(read_rows())
    print(f"Digest: {TOTALS}")
    if ERRORS:
        print(f"{len(ERRORS)} errors")

def process(rows, regions):
    accepted, rejected = parse_rows(rows, regions)
    return {
        "totals": aggregate(accepted),
        "counts": {"read": len(rows), "accepted": len(accepted), "rejected": len(rejected)},
        "rejected": rejected,
    }

Assemble the digest. Write parse_rows, aggregate, and report, and run them over the supplied orders and region lookup. The printed digest must reconcile: read equals accepted plus quarantined.

Loading this exercise…

Optional, and the transfer task for this lesson: make the digest comparable with yesterday's.

Add a second supplied day of orders, produce both digests, and write the three lines that describe the difference between them — orders, units, and value, per group, as a change rather than a level. Then decide what the report should say on the first day, when there is nothing to compare against.

Sign in to track your progress on this exercise.

1. Why should every stage take a value and return a value rather than update shared state?
2. What does the reconciliation equation compare, and where should it be checked?
3. Why should run.py keep domain transformations out of the entry point?
4. Why is it better for aggregate to raise on an unparsed row than to accommodate it?

Sign in to track your progress on this exercise.

One entry point without domain transformations, stages that each take a value and return a value, counts that flow forward and reconcile at the end, settings in one file, and one exit status a scheduler can read. That is the whole digest, and it works on a good day. The next lesson is about the other days: interrupting it deliberately, proving the re-run redoes nothing it already did, and checking that a second run over unchanged input changes nothing at all.