Thirty-five lessons ago you assigned your first variable. Today you build the thing data teams are actually paid to build: a pipeline that takes ugly raw files in one end and pushes a clean, dated, trustworthy report out the other — and doesn't fall over when row 4,217 is garbage. No new syntax in this lesson. Only composition.
What you'll learn
What
An end-to-end pipeline in pure Python: discover input files with pathlib, stream and parse each line, extract fields with regex, convert types defensively, skip-and-count bad rows, aggregate with dictionaries, and write a summary CSV into a dated output folder. Structured as small single-responsibility functions — the shape of every real ETL job.
Why
This is the graduation exercise: every module — variables, strings, control flow, functions, data structures, error handling, files, paths, dates, regex — earns its place in one program. It's also the mental template for everything ahead: pandas, SQL, and ML pipelines all follow this same read → clean → transform → aggregate → write arc, just with bigger engines.
Where it's used
ETL jobs, nightly reports, data-quality monitors, ingestion scripts — and interviews, where 'build a small pipeline' is a standard take-home task for data roles.
Where this runs in production
Airbnb built Airflow to orchestrate thousands of daily read-clean-aggregate-write jobs; each task in those DAGs has exactly the stage structure you're building here.
Results feeds arrive as messy files from thousands of counties; parsing, validating, skip-counting, and aggregating them into live totals is this project's shape under deadline pressure.
Daily settlement files from card networks get parsed, validated row by row (with every rejection logged and counted), aggregated, and written to dated reports — auditability is the product.
Every pipeline decomposes into: DISCOVER (find input files — pathlib glob), EXTRACT (stream lines out of each file — the with-loop), TRANSFORM (line → typed record: regex + conversions + validation), AGGREGATE (records → summary: dict accumulation), LOAD (write outputs — dated paths, separate tree). Naming the stages isn't ceremony: each becomes a function, each function is testable alone, and when production breaks at 2 a.m. you can localize the failure to one stage in minutes.
def parse_line(line: str) -> tuple | None:
"""'2026-07-16 | WIDGET-042 | 3 units | €4.50' -> (date, code, units, price) or None."""
m = LINE_PAT.search(line)
if m is None:
return None
return (
datetime.strptime(m.group(1), '%Y-%m-%d'),
m.group(2).upper(),
int(m.group(3)),
float(m.group(4)),
)The heart of the pipeline is one honest function: string in, typed record or None out. The None contract makes bad data a VALUE the caller counts, not an exception that escapes — the parse attempt is total. Docstring shows a real example; the caller never needs to know regex is inside.
Beginner scripts are bucket brigades: one long main block where water (data) sloshes between untyped variables and one spill soaks everything. A pipeline is a bottling plant: separate STATIONS (functions) — intake inspects crates (discover/extract), the washer rejects cracked bottles into a counted bin (transform + skip-and-count), filling measures precisely (aggregate), labeling stamps the date (load). Each station is inspectable alone, the reject bin is measured — because a plant that silently discards bottles is a plant you can't trust — and the conveyor (main()) just connects stations.
Decide per stage what failure means. A corrupt LINE: skip and count — one bad row must not kill a million-row job. A missing input FOLDER: crash loudly with the path in the message — that's a deployment error, not data noise. The non-negotiable: every skip is counted and printed in the report. '3 rows skipped of 1,204' is data quality information; silent dropping is how a feed that's 40% garbage goes unnoticed for a quarter. This asymmetry — tolerant of rows, strict about structure — is the professional default.
def aggregate(records: list) -> dict:
"""records -> {product_code: total_revenue}"""
totals = {}
for _date, code, units, price in records:
totals[code] = totals.get(code, 0.0) + units * price
return totalsAggregation is the dict-counting idiom scaled up: .get(key, default) + increment. Tuple unpacking names the fields (underscore marks the unused date). Keeping this a pure function — list in, dict out, no prints, no files — makes it trivially testable: feed three records, assert the dict.
The load stage applies the paths lesson verbatim: out_dir = base / 'reports' / run_date.strftime('%Y-%m-%d'), then mkdir(parents=True, exist_ok=True), then write the summary INTO it. Dated folders make history accumulate and runs auditable; the separate output tree makes reruns safe (inputs untouched); mkdir's flags make the job land on a fresh machine without ceremony. Filenames get ISO dates because alphabetical = chronological.
1) One 80-line main block — decompose; if a function needs 'and' to describe, split it. 2) Swallowed skips (except: pass) — count and report, always. 3) Outputs written into the input folder — the rerun-corruption classic. 4) datetime.now() buried in logic — inject the run date; reproducibility dies otherwise. 5) Parsing with split('|') then choking on the one line with a missing pipe — the regex + None contract absorbs shape variation. 6) Aggregating floats without a final round for display. 7) No verification step — end by reading back what you wrote and printing it; trust nothing you didn't check.
Raw files to dated report — each station is one function, each arrow is typed data. Click through the flow.
Build the intake: create a fixture inbox, then discover and stream every line with file/line counts — the pipeline's skeleton before any parsing.
Self-contained fixtures — the habit that makes examples (and tests) runnable anywhere. Note the planted defects: a corrupt line and the 'EUR' price variant. Good pipelines are built AGAINST known-bad data, not clean samples.
Your task
Build a miniature pipeline for expense lines shaped 'YYYY-MM-DD, CATEGORY, €AMOUNT' (amount has 2 decimals). Write parse_line(line) returning (category, float_amount) or None (regex with 2 groups — date matched but not captured; category is uppercase letters). Feed it the lines list, skip-and-count failures, aggregate per-category totals with a dict, and print each 'CATEGORY: total' (sorted, 2 decimals) followed by 'skipped: <n>'.
FOOD: 20.75 TRAVEL: 40.00 skipped: 2
Write your solution in the editor on the right, then hit Run.
What are the five stages of the pipeline arc, in order?
Why does parse_line return None instead of raising on a bad row?
Why does run() take inbox, out_base, and run_date as parameters instead of using Path('inbox') and datetime.now() internally?
What's the root failure?
Write aggregate(records) for records = [('WIDGET', 3, 4.5), ('GIZMO', 1, 12.0), ('WIDGET', 2, 4.5)] returning {code: units*price summed}. Print each 'code: total' sorted, 2 decimals. Expected: GIZMO: 12.00 WIDGET: 22.50
The report lands at reports/<ISO-date>/summary.csv, in a tree separate from inbox/. Which failure does each choice prevent?
Mini end-to-end: lines = ['A-1, 2, 3.00', 'bad', 'B-2, 1, 10.00', 'A-1, 1, 3.00']. Pattern: ([A-Z]-\d+),\s*(\d+),\s*(\d+\.\d{2}) → (code, int qty, float price) or skip-count. Aggregate qty*price per code, print sorted 'code: total' (2 decimals), then 'ok: <n> skipped: <n>'. Expected: A-1: 9.00 B-2: 10.00 ok: 3 skipped: 1
Walk me through designing a script that turns a folder of messy text exports into a daily summary report. What's your structure and why?
I'd structure it as five explicit stages, each a small function. DISCOVER: sorted(inbox.glob('*.txt')) — sorted for run-to-run determinism; a missing inbox raises loudly with the path (deployment error), an empty one is a valid no-op. EXTRACT: stream each file with `with open(..., encoding='utf-8')` and lazy line iteration, so size never matters. TRANSFORM: a parse_line(line) function owning all the intelligence — a compiled regex with capture groups extracts fields, strptime/int/float type them, and the contract is 'typed tuple or None': bad rows become countable values, so the calling loop is a clean if rec is None: skipped += 1 with no exception plumbing. AGGREGATE: a pure function folding records into a dict with .get(key, 0)+increment — list in, dict out, trivially unit-testable. LOAD: outputs into a DATED folder (out_base / run_date ISO), mkdir(parents=True, exist_ok=True), skip counts written INTO the report, then read back and printed as verification. Composition: a run(inbox, out_base, run_date) function with injected dependencies (temp dirs and pinned dates in tests; no hidden now()), under an if __name__ == '__main__' guard. The 'why' running through all of it: localize failures to a stage, make bad data visible rather than fatal, and make every run reproducible and auditable.
How do you decide which errors a pipeline tolerates and which should kill it?
By blast radius and meaning. Row-level dirt — an unparseable line, a bad number, a malformed date — is EXPECTED in real feeds; one row's failure says nothing about the other million, so the policy is skip, count, and report: the parse function returns None (or the loop catches the specific ValueError), a counter increments, and the count ships in the output where humans see it. Structural failures — the input folder missing, a file unreadable, an undecodable encoding, the output disk full — mean the JOB's premises are false; continuing would produce a plausible-looking but wrong report, so these crash immediately and loudly, with the offending path in the message. Two refinements make this production-grade: thresholds — tolerating 0.1% bad rows is resilience, tolerating 60% is negligence, so real jobs fail if skipped/total exceeds a limit (catching upstream format drift the morning it happens); and specificity — except ValueError around the parse, never except Exception, because a broad catch converts genuine bugs (a typo'd variable, a None where a list belonged) into 'skipped rows' and hides them. The anti-pattern to name explicitly: except: pass. Silent partial data is worse than a crash — a crash gets fixed today; quietly missing data gets discovered in an executive meeting.
This pipeline is pure Python. Where does pandas fit, and when would you still write it this way?
pandas industrializes the middle stages: read_csv replaces the manual extract-and-parse for WELL-FORMED delimited data (with dtype and parse_dates doing the typing), .str.extract applies the same regex thinking column-wise, and groupby().sum() is the aggregate stage in one line — the five-stage arc survives intact, each stage just gets a bigger engine. I'd reach for pandas as soon as the data is tabular and fits memory comfortably: less code, fewer bugs, and vectorized speed. Pure Python keeps the job when: the input is NOT table-shaped — ragged log lines, mixed record types, formats needing per-line regex dispatch — where pandas' rectangular model fights you; the file is huge and the aggregation is streaming-friendly (the line loop holds one row in memory; read_csv holds everything, though chunksize is pandas' middle ground); dependencies must be zero (a cron box or minimal container where the standard library is guaranteed — the package-management lesson's concern); or row-level error POLICY is the point — per-line skip-and-count with custom rules is explicit in a loop and awkward mid-read_csv. The honest professional answer: prototype the parse in pure Python to UNDERSTAND the mess, then port to pandas once the data proves tabular — carrying over the same skip-counting and dated-output discipline, because the engineering values are engine-independent.
Common Mistakes to Avoid
1) One monolithic main block — five stages, five functions; testability is the payoff. 2) except: pass — the silent-partial-data machine; skip, COUNT, report. 3) Outputs in the input tree — reruns corrupt sources; separate, dated trees. 4) datetime.now() buried in logic — inject run_date; reproducibility requires it. 5) Broad except Exception around parsing — catches your bugs as 'bad rows'; catch ValueError specifically or use the None contract. 6) No skip threshold — 60% rejects should fail the job, not pad a counter. 7) Skipping verification — read back the report and print it; unverified writes are hope, not engineering. 8) Testing only on clean fixtures — plant corrupt lines in your test data on purpose.
Ask the AI Tutor
Try these prompts in the AI Tutor panel: • 'Review my pipeline against the five-stage arc — where did I merge stages?' • 'Give me three progressively messier sample feeds and critique my parse function on each.' • 'Quiz me: for ten failure scenarios, do I skip-and-count or crash loudly?' • 'Help me add a skip-rate threshold that fails the run above 5%.' • 'Interview mode: ask me the pipeline-design question and grade my structure.'
Glossary
pipeline / ETL — the read → clean → transform → aggregate → write arc (Extract, Transform, Load). stage — one arc step, implemented as one function. parse contract — 'typed record or None': failure as a countable value. skip-and-count — tolerate row dirt, report the tally in the output. skip threshold — the reject rate above which the job fails instead of continuing. dependency injection — passing paths/dates as parameters instead of reaching for globals or now(). idempotent — safe to rerun (separate output trees, exist_ok mkdir). dated outputs — ISO-named run folders that accumulate auditable history. verification — reading back what you wrote before claiming success. entry guard — if __name__ == '__main__': keeping imports side-effect free.
Recommended Resources
• Extend the build: add a skip-rate threshold, a second input format, and a per-date (not just per-product) breakdown — each is a one-function change if your stages are clean, which is the test of whether they are. • Read: the csv module docs (your split/regex parsing, industrial-strength) and a first look at pandas' read_csv page to see these stages with a bigger engine. • Compare: sketch this same pipeline as pandas one-liners and note which stages compress and which (error policy!) don't. • Next in DSM: the Python domain is COMPLETE — 36 lessons from first variable to working pipeline. The road continues into the data stack: pandas DataFrames, where these exact patterns meet columnar power.
Recap
✓ Five stages, five functions: discover (glob) → extract (stream) → transform (parse to typed records) → aggregate (dict fold) → load (dated report). ✓ The parse contract — record or None — turns bad rows into countable values; the loop stays exception-free. ✓ Tolerant of rows, strict about structure: skip-and-count dirt, crash loudly on missing folders — and report every count. ✓ Inject dependencies (paths, run_date); guard the entry point; keep aggregate pure — that's what makes it testable. ✓ Dated, separated, verified outputs: mkdir flags, ISO names, read-back before declaring victory. ✓ The arc is universal — pandas, SQL, and every orchestrator you'll meet run this same shape at scale. Next up: you've completed the Python domain — all 36 lessons. The journey continues in the Data Analysis course with pandas DataFrames, where every idiom you built here gets a columnar engine underneath it.
Run your code to see the output here.