Capstone — Quarry, End to End
In short: nine sections built nine separate pieces of Quarry; none has
run together yet. This capstone is the wiring: one run_quarry(files) that
walks a directory of messy logs, reads each through a retrying ring, decodes
and matches every line, streams the good ones into SQLite, and returns a
report — including everything it couldn't read. The code is mostly written;
your job is the order, which is also where real pipelines break.
Nine pieces, one run: inspector freezes a record into a hashable key,
feed passes records one at a time, ring wraps the reader in
retry/timed, yard finds and decodes files, vault inserts into
SQLite as parameters, extractor turns a line into named fields,
shipment is the package layout below, black box sorts failures,
test rig proves the report true. Every box below is live.
Lay the pieces out as a package
The browser runs one file, so the starter keeps the nine pieces as banners, in package order:
quarry/
├── errors.py QuarryError, SourceError
├── yard.py plant, find_sources, decode, make_reader
├── extractor.py LOG, extract_fields
├── ring.py retry, timed
├── blackbox.py ListHandler, FORMAT
├── vault.py SCHEMA, INSERT
└── cli.py feed, run_quarry, main (this capstone)The dependency rule still holds: cli.py imports everything, nothing
imports cli.py. So run_quarry can be tested with nothing printed, while
main() stays a thin layer that only prints. The
if __name__ == "__main__": guard lets the file run as a script without
doing anything when it's merely imported.
Stream straight into the database
The feed is a generator: it yields one record, then waits. The vault
inserts with executemany. Hand one to the other:
import sqlite3
pulled = []
def rows():
for n in range(3):
pulled.append(n)
yield (n,)
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE t (n INTEGER)")
feed = rows()
print(len(pulled), end=" ")
conn.executemany("INSERT INTO t VALUES (?)", feed)
print(len(pulled))executemany pulls any iterable one row at a time, so Quarry never holds a
file's records in a list — each line is matched, checked for duplicates and
inserted before the next is even read:
The trace callback prints each statement as SQLite runs it, and the feed's
own prints land between the inserts — a million-line log costs the same
memory as three lines. So every tally the feed keeps (skipped,
duplicates, failed) is only final once executemany returns — read it
after the with conn: block, never before.
Wrap the reader before it can lie
make_reader(root) returns read_source(path): a disk on a bad day. A
flaky- file fails its first read, a down- file fails every read, both
raising SourceError. The ring's retry(3) and timed(timings) wrap it —
and the order you apply them in decides what timed measures:
calls = []
def timed(func):
def wrapper(*args):
calls.append(func.__name__)
return func(*args)
return wrapper
def retry(times):
def decorate(func):
def wrapper(*args):
for attempt in range(times):
try:
return func(*args)
except OSError:
pass
raise OSError("gave up")
return wrapper
return decorate
failures = [1]
def read(path):
if failures:
failures.pop()
raise OSError(path)
return b"rows"
read = retry(3)(timed(read))
read("flaky-mount.log")
print(len(calls))Quarry wants retry(3)(timed(timings)(make_reader(root))) — timed inside —
so len(timings["read_source"]) counts attempts, showing a mount
getting worse before it fails outright. Retry also logs each failed attempt
on quarry.ring, so the report shows which files needed a second try.
Follow one line through every stage
A single line from a Windows machine, stepping through bytes, a guessed
encoding, a regex match and a hashable key. Watch encoding settle on
cp1252 after UTF-8 refuses the é:
splitlines() drops the \r\n, so a Windows line matches the same pattern
as a Unix one, and fullmatch — not search — refuses a line with junk in
front. The key is a tuple: a dict can't go in a set, but its values,
frozen, can; it leaves out source on purpose, so the same line in
app.log and archive/app.log counts as one event. {"source": ..., **fields} has exactly the names INSERT uses as placeholders, so a message
like '); DROP TABLE records; -- goes in as a parameter, stored as text.
Sort trouble into three buckets
| trouble | example | what the run does | logged as |
|---|---|---|---|
| a source fails | down-nas.log never reads |
retry twice, give up on that file, move on | ERROR quarry.yard: gave up on down-nas.log |
| a line is bad | not a log line |
skip it, count it, keep going | WARNING quarry.extractor: skipped app.log:3 |
| Quarry has a bug | a KeyError in your own code |
nothing — let it crash with its traceback |
Only SourceError is caught, and only around the read — catching
Exception anywhere would turn a real bug into a healthy-looking report.
The run attaches a ListHandler (level WARNING) to the quarry logger;
quarry.yard, quarry.extractor and quarry.ring all propagate up to it,
so one handler hears all three. It has to come off when the run ends — this
version forgets:
The first run's report should hold only its own warning, but the second run's warning lands in it too. Make each run clean up after itself.
The grader calls run_quarry five times in one process; without the
finally, the fifth report would carry all five runs' warnings.
Prove it before you trust it
A report is only as good as its tests. A small self-contained runner, two tests of the extractor:
The most useful tests are about the drawing's dashed arrows:
report["failed"] == ["down-nas.log"] for a dead mount, and
report["duplicates"] == 1 when archive/app.log repeats app.log — a run
that silently drops a file looks just like one with nothing to read.
Cross-question. Why is
feeda separate generator, not a loop insiderun_quarry? It holds every decision — retry or give up, skip, dedupe — so it tests alone:list(feed(root, read, tally)), no database needed.
Quick practice
seen = set()
results = []
for fields in [{"level": "INFO", "msg": "ok"}, {"msg": "ok", "level": "INFO"}]:
key = tuple(sorted(fields.items()))
results.append(key in seen)
seen.add(key)
print(results)import sqlite3
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE t (level TEXT)")
conn.executemany("INSERT INTO t VALUES (?)", [("WARN",), ("ERROR",), ("WARN",)])
print(conn.execute("SELECT level, COUNT(*) FROM t GROUP BY level ORDER BY level").fetchall())🎯 Your turn
The starter holds every finished piece. You write two functions in cli.py:
feed(root, read, tally), a generator. For each path from
find_sources(root), with rel its path relative to root:
read(path). OnSourceError, logYARD.error("gave up on %s", rel), appendreltotally["failed"]and move on.decodethe bytes; recordtally["sources"][rel] = encoding.- Number the lines from 1, skip blanks. If
extract_fieldsreturnsNone, add 1 totally["skipped"]and logEXTRACTOR.warning("skipped %s:%d", rel, number). - If the fields were seen before (any file), add 1 to
tally["duplicates"]; otherwise yield{"source": rel, **fields}.
run_quarry(files): attach a ListHandler to
logging.getLogger("quarry") (removed in finally), open a :memory:
database, plant the files in a TemporaryDirectory, build
read = retry(3)(timed(timings)(make_reader(root))), and
conn.executemany(INSERT, feed(root, read, tally)) in one transaction.
Return:
{"sources": {...}, "failed": [...], "stored": COUNT(*), "skipped": n,
"duplicates": n, "levels": {level: count}, "attempts": len(timings["read_source"]),
"log": box.lines}Example: app.log holding one INFO [yard] started line gives
{"sources": {"app.log": "utf-8"}, "failed": [], "stored": 1, "skipped": 0, "duplicates": 0, "levels": {"INFO": 1}, "attempts": 1, "log": []}.
Press ▶ Run to see main() print a report, then ✓ Check. Green
means Quarry runs end to end: messy files in, a trustworthy report out. ✅