Python lets a program do almost anything from anywhere: change a global, read a file halfway through a calculation, shuffle with whatever seed the process happens to hold. The Sean Dinwiddie’s Webmastery team gives most of that freedom up, gladly, for something better: the same inputs give the same outputs on every machine, and a failing test can say exactly which rule broke.
Eight habits get it there. Each comes with a small example that runs on the standard library and pytest. From here on the module works in Python, because the libraries that train and measure a model are written for Python first, as the Module 6 opener sets out. Every boundary it crosses is held to a contract the way this page holds its training rows, from the config and the design files in the module’s shared format to a model’s output.
One layout, run as modules
Code is grouped by what it does, one package per domain, and nothing sits loose at the top of the repository. Configuration and data live beside the package, never inside it:
pyproject.toml what the project uses, and its Python
uv.lock every package at an exact version, resolved from it
pytest.ini
config/train.json every value a run depends on
data/
pipeline/
config/read.py
data/catalog_rows.py
data/contract.py
evaluation/
release/
tests/Every entry point runs as a module from the repository’s root, so its imports are absolute and resolve the same way from a terminal, a test or a scheduled job. An entry point’s main() returns an exit code, and the module hands it to the shell:
# pipeline/data/catalog_rows.py
"""Print training rows as JSON lines, one per row."""
from __future__ import annotations
import json
import sys
from pipeline.data.contexts import CATALOG
from pipeline.data.order import rotate
def catalog_rows(count: int) -> tuple[dict[str, str], ...]:
return tuple(
{"id": f"row-{index:03d}", "place": rotate(CATALOG, index).place}
for index in range(count)
)
def main(argv: list[str] | None = None) -> int:
args = sys.argv[1:] if argv is None else argv
count = int(args[0]) if args else 4
for row in catalog_rows(count):
print(json.dumps(row, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())$ python -m pipeline.data.catalog_rows
{"id": "row-000", "place": "the north trail"}
{"id": "row-001", "place": "the east parking lot"}
{"id": "row-002", "place": "the library annex"}
{"id": "row-003", "place": "the north trail"}Run it as a file instead, and the same code falls over. Python puts the file’s own folder first on the import path, not the root:
$ python pipeline/data/catalog_rows.py 2>&1 | tail -1
ModuleNotFoundError: No module named 'pipeline'python -m it is the current directory, here the root, so import pipeline finds the package (the violet line). Run as a file, it is the file’s own folder, data/, and there is no pipeline/ in there.A locked environment
Each project has its own virtual environment, made from a lock. People edit pyproject.toml: what the project uses, the Python it runs on, and a date after which nothing new is taken:
[project]
name = "trail-notices"
version = "0.1.0"
requires-python = "==3.11.*"
dependencies = []
[dependency-groups]
dev = ["pytest"]
[tool.uv]
exclude-newer = "2026-09-01T00:00:00Z"uv resolves it into uv.lock: every package and every dependency at an exact version, with a hash for each file and nothing uploaded after that date, for every platform the project allows. Lock it again next month and you get the same lock.
The environment is made from the lock alone. uv sync --locked installs exactly what the lock names, and removes anything it doesn’t:
$ uv sync --locked
Using CPython 3.11.15 interpreter at: /usr/local/bin/python3
Creating virtual environment at: .venv
Resolved 7 packages in 2ms
Installed 5 packages in 7ms
+ iniconfig==2.3.0
+ packaging==26.3
+ pluggy==1.6.0
+ pygments==2.21.0
+ pytest==9.1.1--locked means the lock must already match pyproject.toml, and uv never rewrites it on the way. On its own, uv lock --locked is the check:
$ uv lock --locked
Resolved 7 packages in 3msAdd six to the dependencies by hand, without locking again, and the same check stops the run before anything installs it:
$ uv lock --locked; echo "exit $?"
Resolved 8 packages in 9ms
The lockfile at `uv.lock` needs to be updated, but `--locked` was provided. To update the lockfile, run `uv lock`.
exit 1Pure functions, the same answer every time
A function that reads only its arguments and returns a new value can be tested with plain data and trusted on any machine. Data generation holds to that strictly.
A row’s content is chosen by its index, turning through a fixed catalog like a dial; nothing is drawn at random. Where an order has to be shuffled, the shuffle gets a generator of its own, seeded from the run’s settings. The global one belongs to every library in the process, and any of them can disturb it:
# pipeline/data/order.py
"""Deterministic choices: the same inputs give the same rows on every machine."""
from __future__ import annotations
import random
from typing import Sequence, TypeVar
T = TypeVar("T")
def rotate(items: Sequence[T], index: int) -> T:
"""The item for row `index`: a fixed rotation, never a random draw."""
return items[index % len(items)]
def seeded_order(count: int, seed: int) -> tuple[int, ...]:
"""Row indices in a shuffled order that the seed fixes."""
rng = random.Random(seed) # its own generator, not the global one
indices = list(range(count))
rng.shuffle(indices)
return tuple(indices)The hash() trap
For strings, Python’s built-in hash() is salted afresh in every process, so anything derived from it changes from run to run. The seeded order doesn’t budge:
$ for run in 1 2 3; do python -c "print(hash('north trail') % 1000)"; done
928
563
205
$ for run in 1 2 3; do python -c "from pipeline.data.order import seeded_order; print(seeded_order(8, seed=11))"; done
(1, 0, 2, 5, 3, 4, 6, 7)
(1, 0, 2, 5, 3, 4, 6, 7)
(1, 0, 2, 5, 3, 4, 6, 7)A little housekeeping
Three small rules let two runs be compared line by line:
- output is sorted;
- JSON is written with sorted keys;
- floats are rounded before they are compared.
Heavy libraries are imported inside the functions at the edge that need them, so a test of the core never loads a GPU stack.
The lecture What Is a Function? sets out what makes a function pure, and Functional Programming in Other Languages draws the line between the pure core and the shell around it.
Frozen data
Records are frozen dataclasses. A frozen record can’t change after it is made, so a value passed into a function is the same value when the function returns. A change is a new record, made with dataclasses.replace:
# pipeline/data/contexts.py
"""The made-up situations that training rows are written about."""
from __future__ import annotations
from dataclasses import dataclass
@dataclass(frozen=True)
class Context:
place: str
state: str
reason: str
CATALOG = (
Context("the north trail", "closed until Friday", "a washed-out bridge"),
Context("the east parking lot", "open from 7 in the morning", "new gate hours"),
Context("the library annex", "closed on Mondays", "a change in staffing"),
)from dataclasses import FrozenInstanceError, replace
from pipeline.data.contexts import CATALOG
trail = CATALOG[0]
try:
trail.state = "open"
except FrozenInstanceError as error:
print("refused:", error)
reopened = replace(trail, state="open")
print(reopened)
print(trail)refused: cannot assign to field 'state'
Context(place='the north trail', state='open', reason='a washed-out bridge')
Context(place='the north trail', state='closed until Friday', reason='a washed-out bridge')The trail in the catalog is still closed until Friday; the reopened one is a new record beside it.
One config, read once
Every value a run depends on sits in one file, config/train.json: the model and its pinned revision, the data files, the seed, the training settings. It is short enough to review at a glance, and changing a run means changing that file:
{
"model": "example/base-model",
"revision": "pinned",
"train": "data/train.jsonl",
"validation": "data/validation.jsonl",
"seed": 11,
"epochs": 2
}A run reads it once, at its entry point, into a frozen record, and hands the record to whatever needs it. Nothing deeper opens the file again, so every part of the run sees the same values.
Reading checks every key against the record’s fields and reports every fault at once, the way the next section’s contract does:
# pipeline/config/read.py
"""Read the run's config once, at its entry point, into a frozen record."""
from __future__ import annotations
import json
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any, get_type_hints
@dataclass(frozen=True)
class Config:
model: str
revision: str
train: str
validation: str
seed: int
epochs: int
def config_errors(raw: Any) -> tuple[str, ...]:
"""Every way the file differs from Config, all in one report."""
if not isinstance(raw, dict):
return ("a config is one JSON object",)
kinds = get_type_hints(Config)
return (
*(f"unknown key {key!r}" for key in sorted(set(raw) - set(kinds))),
*(f"missing key {key!r}" for key in kinds if key not in raw),
*(
f"{key} should be {kind.__name__}, found {type(raw[key]).__name__}"
for key, kind in kinds.items()
if key in raw and type(raw[key]) is not kind
),
)
def read_config(path: Path) -> Config:
raw = json.loads(path.read_text(encoding="utf-8"))
errors = config_errors(raw)
if errors:
raise ValueError("\n".join(errors))
return Config(**raw)
def main(argv: list[str] | None = None) -> int:
args = sys.argv[1:] if argv is None else argv
print(read_config(Path(args[0])))
return 0
if __name__ == "__main__":
raise SystemExit(main())$ python -m pipeline.config.read config/train.json
Config(model='example/base-model', revision='pinned', train='data/train.jsonl', validation='data/validation.jsonl', seed=11, epochs=2)A draft with a stray key, a missing one and a seed written as text gets all three back in one report:
from pipeline.config.read import config_errors
draft = {
"model": "example/base-model", "revision": "pinned", "train": "data/train.jsonl",
"seed": "11", "epochs": 2, "learning_rate": 0.0002,
}
for error in config_errors(draft):
print(error)unknown key 'learning_rate'
missing key 'validation'
seed should be int, found strThe lecture Practical Applications of Functional Programming freezes its configuration too, with caller overrides layered over defaults in an explicit order. Here the file holds every value, so there is nothing to layer.
A design file in the format Geometric reasoning as data sets out is read the same way: once, by a reader of its own, from the one place it lives, the way Geometric reasoning in model training reads data/trail.design.json.
Contracts that report every fault
Data crosses a boundary only after it is checked against its contract:
- the keys a row must carry, and the one it may carry (the oracle, which Geometric reasoning in model training gives every row it writes itself);
- the allowed values;
- the rules between fields.
The checks are independent of each other, so all of them run, and the validator raises once with every fault, not just the first. A rule about a value judges only a key the row has, so a missing key is reported once, as missing. Whoever fixes the data file gets the whole list in one go.
Counts and other derived facts are recomputed from the data, never trusted as declared:
# pipeline/data/contract.py
"""Check training rows against their contract and report every fault at once."""
from __future__ import annotations
import json
import sys
from collections import Counter
from pathlib import Path
from typing import Any, Iterable, Mapping
REQUIRED_KEYS = frozenset({"id", "split", "prompt", "chosen", "rejected"})
OPTIONAL_KEYS = frozenset({"oracle"}) # what its answers must and must not say
SPLITS = frozenset({"train", "validation", "test"})
def row_errors(row: Mapping[str, Any]) -> tuple[str, ...]:
keys = set(row)
rules = ( # each rule and what to say when it doesn't hold
(keys <= REQUIRED_KEYS | OPTIONAL_KEYS, f"unexpected keys {sorted(keys - REQUIRED_KEYS - OPTIONAL_KEYS)}"),
(keys >= REQUIRED_KEYS, f"missing keys {sorted(REQUIRED_KEYS - keys)}"),
("split" not in row or row["split"] in SPLITS, f"unknown split {row.get('split')!r}"),
("prompt" not in row or bool(str(row["prompt"]).strip()), "empty prompt"),
(not keys >= {"chosen", "rejected"} or row["chosen"] != row["rejected"], "chosen and rejected are the same"),
)
return tuple(message for holds, message in rules if not holds)
def count_errors(declared: Mapping[str, int], rows: Iterable[Mapping[str, Any]]) -> tuple[str, ...]:
"""Recount the rows in each split rather than trusting the declared counts."""
found = Counter(row.get("split") for row in rows)
return tuple(
f"{split}: declared {declared.get(split, 0)}, found {found.get(split, 0)}"
for split in sorted(set(declared) | set(found), key=str)
if declared.get(split, 0) != found.get(split, 0)
)
def validate(rows: Iterable[Mapping[str, Any]], declared: Mapping[str, int]) -> None:
rows = tuple(rows)
errors = (
*(f"{row.get('id', '?')}: {error}" for row in rows for error in row_errors(row)),
*count_errors(declared, rows),
)
if errors:
raise ValueError("\n".join(errors))
def main(argv: list[str] | None = None) -> int:
rows_path, counts_path = sys.argv[1:] if argv is None else argv
rows = [json.loads(line) for line in Path(rows_path).read_text().splitlines() if line]
validate(rows, json.loads(Path(counts_path).read_text()))
print(f"{len(rows)} rows match their contract")
return 0
if __name__ == "__main__":
raise SystemExit(main())from pipeline.data.contract import validate
rows = (
{"id": "row-000", "split": "train", "prompt": "Is the north trail open?",
"chosen": "It is closed until Friday.", "rejected": "Closed till Friday!"},
{"id": "row-001", "split": "trian", "prompt": " ", "note": "draft",
"chosen": "Open from 7 a.m.", "rejected": "Open from 7 a.m."},
)
try:
validate(rows, declared={"train": 2})
except ValueError as error:
print(error)row-001: unexpected keys ['note']
row-001: unknown split 'trian'
row-001: empty prompt
row-001: chosen and rejected are the same
train: declared 2, found 1
trian: declared 0, found 1The lecture Practical Applications of Functional Programming accumulates errors the same way in TypeScript.
Tests named for the behavior
Each test’s name states the rule it proves, so a failing run reads as a list of broken rules.
A contract is tested by breaking a valid row one way at a time and checking that each break is refused, by name. Several breaks at once have to come back in one error:
# tests/test_contract.py
import pytest
from pipeline.data.contract import validate
VALID = {
"id": "row-000",
"split": "train",
"prompt": "Is the north trail open?",
"chosen": "It is closed until Friday.",
"rejected": "Closed till Friday!",
}
BROKEN = (
("unknown-split", {"split": "trian"}, "unknown split 'trian'"),
("empty-prompt", {"prompt": " "}, "empty prompt"),
("same-answers", {"rejected": VALID["chosen"]}, "chosen and rejected are the same"),
("extra-key", {"note": "draft"}, r"unexpected keys \['note'\]"),
)
def test_a_valid_row_passes():
validate((VALID,), declared={"train": 1})
@pytest.mark.parametrize("change, error", [case[1:] for case in BROKEN], ids=[case[0] for case in BROKEN])
def test_each_broken_row_is_refused_by_name(change, error):
with pytest.raises(ValueError, match=error):
validate(({**VALID, **change},), declared={"train": 1})
def test_a_wrong_declared_count_is_refused():
with pytest.raises(ValueError, match="train: declared 2, found 1"):
validate((VALID,), declared={"train": 2})
def test_every_fault_is_reported_in_one_error():
with pytest.raises(ValueError) as raised:
validate(({**VALID, "split": "trian", "prompt": ""},), declared={"train": 1})
assert len(str(raised.value).splitlines()) == 4A test that has to fail
A test that has never failed has proved nothing, so a test can break the code on purpose. Here one test swaps in a broken rotation and expects the rotation test to fail. The last two hold the shuffled order to its seed:
# tests/test_rows.py
import pytest
import pipeline.data.catalog_rows as build
from pipeline.data.contexts import CATALOG
from pipeline.data.order import seeded_order
def test_rows_rotate_through_the_catalog():
places = [row["place"] for row in build.catalog_rows(4)]
assert places == [context.place for context in (*CATALOG, CATALOG[0])]
def test_the_rotation_test_fails_when_the_rotation_breaks(monkeypatch):
monkeypatch.setattr(build, "rotate", lambda items, index: items[0]) # broken on purpose
with pytest.raises(AssertionError):
test_rows_rotate_through_the_catalog()
def test_the_same_seed_gives_the_same_order():
assert seeded_order(8, seed=11) == seeded_order(8, seed=11)
def test_a_new_seed_gives_a_new_order_of_the_same_rows():
first, second = seeded_order(8, seed=11), seeded_order(8, seed=12)
assert first != second and sorted(first) == sorted(second)The run itself
Every run reads its settings from pytest.ini at the root: where the tests are, no cache left behind, and the classic output without a progress percentage:
[pytest]
testpaths = tests
addopts = -p no:cacheprovider
console_output_style = classic$ python -m pytest -v --no-header tests/test_contract.py tests/test_rows.py
============================= test session starts ==============================
collecting ... collected 11 items
tests/test_contract.py::test_a_valid_row_passes PASSED
tests/test_contract.py::test_each_broken_row_is_refused_by_name[unknown-split] PASSED
tests/test_contract.py::test_each_broken_row_is_refused_by_name[empty-prompt] PASSED
tests/test_contract.py::test_each_broken_row_is_refused_by_name[same-answers] PASSED
tests/test_contract.py::test_each_broken_row_is_refused_by_name[extra-key] PASSED
tests/test_contract.py::test_a_wrong_declared_count_is_refused PASSED
tests/test_contract.py::test_every_fault_is_reported_in_one_error PASSED
tests/test_rows.py::test_rows_rotate_through_the_catalog PASSED
tests/test_rows.py::test_the_rotation_test_fails_when_the_rotation_breaks PASSED
tests/test_rows.py::test_the_same_seed_gives_the_same_order PASSED
tests/test_rows.py::test_a_new_seed_gives_a_new_order_of_the_same_rows PASSED
============================== 11 passed in 0.02s ==============================A skip means a test didn’t run, and a green run hides it. So conftest.py fails any run with a skip:
# tests/conftest.py
"""A run with a skip in it fails: the skipped test never ran."""
def skipped(config):
reporter = config.pluginmanager.get_plugin("terminalreporter")
return len(reporter.stats.get("skipped", ())) if reporter else 0
def pytest_terminal_summary(terminalreporter, config):
if skipped(config):
terminalreporter.write_line(f"{skipped(config)} skipped: every test must run")
def pytest_sessionfinish(session):
if skipped(session.config):
session.exitstatus = 1The whole suite holds 17 tests: these eleven, four in tests/test_config.py for the config reader (the frozen record, every fault in one report, True refused as a number, and a config that isn’t one object), and the two in tests/test_inputs.py further down. With a skipped test added, the run reports it and exits non-zero:
$ python -m pytest -q; echo "exit $?"
.............s....
1 skipped: every test must run
17 passed, 1 skipped in 0.03s
exit 1Try it: four faults, and one the contract can’t see
Work in a copy of the layout, so the runs further down still print what this page shows. The data file data/practice.jsonl holds six rows and four faults the contract can see, each against a different rule, and data/practice-counts.json declares the six: {"train": 6}.
{"id": "row-000", "split": "train", "prompt": "Is the north trail open?", "chosen": "It is closed until Friday, because of a washed-out bridge.", "rejected": "Closed till Friday!"}
{"id": "row-001", "split": "train", "prompt": "Why is the north trail closed?", "chosen": "A washed-out bridge. It reopens on Friday.", "rejected": "Ask a ranger.", "source": "posted notice"}
{"id": "row-002", "split": "train", "prompt": "When does the east parking lot open?", "chosen": "At 7 in the morning, under the new gate hours."}
{"id": "row-003", "split": "train", "prompt": "", "chosen": "Yes, from 7 in the morning.", "rejected": "Yes."}
{"id": "row-004", "split": "train", "prompt": "Is the library annex open on Mondays?", "chosen": "No. It is closed on Mondays, because of a change in staffing.", "rejected": "No. It is closed on Mondays, because of a change in staffing."}
{"id": "row-000", "split": "train", "prompt": "Which days is the library annex open?", "chosen": "Every day but Monday, because of a change in staffing.", "rejected": "Most days."}Run the contract once, and fix all four from its report:
$ python -m pipeline.data.contract data/practice.jsonl data/practice-counts.json 2>&1 | tail -4
ValueError: row-001: unexpected keys ['source']
row-002: missing keys ['rejected']
row-003: empty prompt
row-004: chosen and rejected are the sameOne line per fault, each naming its row and the rule it broke, so every fix comes straight from the report:
row-001: dropsource(if every row should carry one, that is a change to the contract, made on purpose);row-002: add the rejected answer it lacks;row-003: write the question its answers reply to;row-004: write a rejected answer that differs from the chosen one.
Then run it again:
$ python -m pipeline.data.contract data/practice.jsonl data/practice-counts.json
6 rows match their contractIt passes, and it still holds a fault. The first and last rows are both row-000: the last was copied from the first and edited, and the id came along. Nothing in the contract compares ids across rows, so a report about row-000 would point at two rows. That fifth fault needs a new rule: each id names one row.
Its test comes first, named for the rule:
# tests/test_contract.py, one test added
def test_a_repeated_id_is_refused():
copy = {**VALID, "prompt": "Which days is the library annex open?"}
with pytest.raises(ValueError, match="row-000: id used by 2 rows"):
validate((VALID, copy), declared={"train": 2})Run it before the rule exists. It has to fail, and it does, by name:
$ python -m pytest -q tests/test_contract.py
.......F
=================================== FAILURES ===================================
________________________ test_a_repeated_id_is_refused _________________________
def test_a_repeated_id_is_refused():
copy = {**VALID, "prompt": "Which days is the library annex open?"}
> with pytest.raises(ValueError, match="row-000: id used by 2 rows"):
E Failed: DID NOT RAISE ValueError
tests/test_contract.py:45: Failed
=========================== short test summary info ============================
FAILED tests/test_contract.py::test_a_repeated_id_is_refused - Failed: DID NO...
1 failed, 7 passed in 0.02sThen the rule, beside count_errors, and one line in validate that runs it with the others. It counts only rows that have an id: a row without one is the missing-keys rule’s to report, so no fault is reported twice.
# pipeline/data/contract.py, with the new rule added
def id_errors(rows: Iterable[Mapping[str, Any]]) -> tuple[str, ...]:
"""Each id names one row, so every line of a report points at one row."""
found = Counter(row["id"] for row in rows if "id" in row)
return tuple(
f"{row_id}: id used by {found[row_id]} rows"
for row_id in sorted(found, key=str)
if found[row_id] > 1
)
def validate(rows: Iterable[Mapping[str, Any]], declared: Mapping[str, int]) -> None:
rows = tuple(rows)
errors = (
*(f"{row.get('id', '?')}: {error}" for row in rows for error in row_errors(row)),
*count_errors(declared, rows),
*id_errors(rows),
)
if errors:
raise ValueError("\n".join(errors))Now the test passes, and the practice file’s report names the repeated id:
$ python -m pytest -q tests/test_contract.py
........
8 passed in 0.01s
$ python -m pipeline.data.contract data/practice.jsonl data/practice-counts.json 2>&1 | tail -1
ValueError: row-000: id used by 2 rowsGive the last row its own id, row-005, and the file passes the contract, new rule and all:
$ python -m pipeline.data.contract data/practice.jsonl data/practice-counts.json
6 rows match their contractOne command checks everything
The config and the data each get a test, so the test run checks them with everything else:
# tests/test_inputs.py
"""The run's own inputs meet their contracts: the config reads, and the rows pass."""
import json
from pathlib import Path
from pipeline.config.read import read_config
from pipeline.data.contract import validate
ROOT = Path(__file__).resolve().parents[1]
def test_the_config_reads():
read_config(ROOT / "config/train.json")
def test_the_training_rows_the_config_names_meet_their_contract():
config = read_config(ROOT / "config/train.json")
rows = [json.loads(line) for line in (ROOT / config.train).read_text().splitlines() if line]
validate(rows, json.loads((ROOT / "data/counts.json").read_text()))data/train.jsonl holds six preference rows that meet the contract, and data/counts.json declares them: {"train": 6}. Each row asks for a brief, formal, cautious voice; its chosen answer keeps it, and its rejected answer misses by one trait.
uv run --locked --exact runs a command in the project’s environment. It stops if the lock no longer matches pyproject.toml, brings the environment exactly in line with the lock, removing anything the lock doesn’t name, and only then runs the command. So one command checks, before any training run:
- the lock is current;
- the config reads;
- the training rows the config names meet their contract;
- every test passes.
$ uv run --locked --exact python -m pytest -q
.................
17 passed in 0.02sIt never writes the lock, the config or the data. Regenerating data is a separate command, so checking can never change what it checks.
Next, these habits go to work before any training starts: Measuring reasoning on the geometry writes the measurement first, the way a scenario comes before its code.