Checkpoint: Levels Everywhere

Deciding to use Level was the easy part. The analyzer has levels in four places: the parser makes them, the summary counts them, the filters compare them, and the store saves and loads them. Change the type in one place and every other place has to agree.

This is the most common kind of change in a real codebase. It’s not a new feature. It’s one idea, changed everywhere it lives. The way through it is the one from Day 1: small steps, with the tests telling you where you are.

Day 5 checkpoint

Exercise

Make the analyzer use Level

The playground holds Day 4’s analyzer.py, filters.py and store.py, where levels are strings, plus levels.py with the Level enum from the last lesson. The tests have already been updated to expect Level everywhere. Right now seven fail.

Make them all pass. Work in this order, running the tests after each step:

analyzer.py. parse_record should turn the level’s text into a Level with Level.from_text, and report unknown levels with the same ParseError as before. LEVELS can go. Then look at summarize: which levels count as errors now?
filters.py. level_at_least should accept a Level or a level’s name, and compare Levels. LEVEL_ORDER can go.
store.py. Once steps 1 and 2 pass, run the tests again. Four store tests that were passing now fail. Read what the database stored, and fix the way levels go in and come out.

"""Log analyzer, Day 4: parsing and counting, configured by the functions you pass in.""" import re from collections import Counter from dataclasses import dataclass, field from datetime import datetime LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"} # key=value, where a value is either "quoted, spaces allowed" or a single word. FIELD = re.compile(r'(?P<key>\w+)=(?P<value>"[^"]*"|\S+)') @dataclass(frozen=True) class LogRecord: timestamp: datetime level: str source: str message: str fields: dict[str, str] = field(default_factory=dict) line_number: int = 0 class ParseError(ValueError): """A line that is not a valid log record.""" def __init__(self, line_number, reason, line): super().__init__(f"line {line_number}: {reason}: {line!r}") self.line_number = line_number self.reason = reason self.line = line def parse_fields(text): """Turn 'a=1 b="two words"' into {'a': '1', 'b': 'two words'}.""" return {m["key"]: m["value"].strip('"') for m in FIELD.finditer(text)} def parse_record(line, line_number=0): """Parse one line, or raise ParseError saying what is wrong with it.""" parts = line.split(maxsplit=3) if len(parts) < 4: raise ParseError(line_number, "too few parts", line) date, time, level, rest = parts try: timestamp = datetime.fromisoformat(f"{date} {time}") except ValueError: raise ParseError(line_number, "bad timestamp", line) from None if level not in LEVELS: raise ParseError(line_number, f"unknown level {level!r}", line) source, colon, message = rest.partition(":") if not colon: raise ParseError(line_number, "missing 'source:'", line) message = message.strip() return LogRecord(timestamp, level, source, message, parse_fields(message), line_number) def read_lines(path): """Yield the lines of a file one at a time, without line endings.""" with open(path) as file: for line in file: yield line.rstrip("\n") def parse_lines(lines, on_error=None): """Yield a record per good line. A bad line's ParseError is passed to on_error. With no on_error it is raised, so by default the first bad line stops everything: nothing is dropped silently. """ for number, line in enumerate(lines, start=1): if not line.strip(): continue try: record = parse_record(line, number) except ParseError as error: if on_error is None: raise on_error(error) else: yield record def count(records, key, keep=None): """Count records by key(record), counting only those where keep(record) is true.""" return Counter(key(r) for r in records if keep is None or keep(r)) def summarize(records): """Count levels and errors per source in a single pass.""" levels, error_sources = Counter(), Counter() for record in records: levels[record.level] += 1 if record.level == "ERROR": error_sources[record.source] += 1 return levels, error_sources

"""Ways to choose records: each function builds a predicate, a yes/no question about one record.""" LEVEL_ORDER = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] def level_at_least(minimum): """Keep records at minimum or more severe: level_at_least("WARNING").""" rank = LEVEL_ORDER.index(minimum) # an unknown level fails now, not later def keep(record): return LEVEL_ORDER.index(record.level) >= rank return keep def source_is(*sources): """Keep records from any of the given sources.""" wanted = set(sources) def keep(record): return record.source in wanted return keep def between(start, end): """Keep records from start (included) up to end (not included).""" def keep(record): return start <= record.timestamp < end return keep def all_of(*predicates): """Keep records that every predicate keeps.""" def keep(record): return all(predicate(record) for predicate in predicates) return keep

"""Log analyzer, Day 4: keep parsed records in SQLite, and ask questions in SQL.""" import json import sqlite3 from contextlib import contextmanager from datetime import datetime from analyzer import LogRecord, parse_fields SCHEMA = """ CREATE TABLE IF NOT EXISTS records ( log TEXT NOT NULL, line_number INTEGER NOT NULL, timestamp TEXT NOT NULL, -- ISO 8601, so text order is time order level TEXT NOT NULL, source TEXT NOT NULL, message TEXT NOT NULL, PRIMARY KEY (log, line_number) ) """ def connect(path=":memory:"): """Open (or create) a database with the records table in it.""" conn = sqlite3.connect(path) conn.row_factory = sqlite3.Row conn.execute(SCHEMA) return conn @contextmanager def open_store(path=":memory:"): """Connect, and close the connection when the with block ends, however it ends.""" conn = connect(path) try: yield conn finally: conn.close() def save_records(conn, log, records): """Store records read from one log. Lines already stored are skipped. Returns how many new records were stored. """ rows = ((log, r.line_number, r.timestamp.isoformat(sep=" "), r.level, r.source, r.message) for r in records) with conn: # one transaction: all rows, or none cursor = conn.executemany( "INSERT INTO records VALUES (?, ?, ?, ?, ?, ?) " "ON CONFLICT (log, line_number) DO NOTHING", rows) return cursor.rowcount def record_from_row(row): """Rebuild a LogRecord. Fields are re-parsed from the message, not stored twice.""" return LogRecord( timestamp=datetime.fromisoformat(row["timestamp"]), level=row["level"], source=row["source"], message=row["message"], fields=parse_fields(row["message"]), line_number=row["line_number"], ) def load_records(conn, log): rows = conn.execute( "SELECT * FROM records WHERE log = ? ORDER BY line_number", (log,)) return [record_from_row(row) for row in rows] def search(conn, text): """Records whose message contains text, oldest first.""" rows = conn.execute( "SELECT * FROM records WHERE message LIKE ? ORDER BY timestamp", (f"%{text}%",)) return [record_from_row(row) for row in rows] COUNTABLE = {"level", "source"} def count_by(conn, column, level=None): """Count records per level or per source, busiest first, optionally for one level.""" if column not in COUNTABLE: # a column name can't be a ? parameter raise ValueError(f"can't count by {column!r}; choose from {sorted(COUNTABLE)}") where, params = ("WHERE level = ?", (level,)) if level else ("", ()) rows = conn.execute( f"SELECT {column} AS key, COUNT(*) AS n FROM records {where} " f"GROUP BY {column} ORDER BY n DESC, key", params) return {row["key"]: row["n"] for row in rows} def summary(conn): """Everything the daily report needs, as plain JSON-ready values.""" total, first, last = conn.execute( "SELECT COUNT(*), MIN(timestamp), MAX(timestamp) FROM records").fetchone() hours = conn.execute( "SELECT CAST(substr(timestamp, 12, 2) AS INTEGER) AS hour, COUNT(*) AS n " "FROM records WHERE level = 'ERROR' GROUP BY hour ORDER BY hour") return { "records": total, "first": first, "last": last, "levels": count_by(conn, "level"), "errors_by_source": count_by(conn, "source", level="ERROR"), "errors_by_hour": [{"hour": row["hour"], "errors": row["n"]} for row in hours], } def save_summary(data, path): with open(path, "w") as file: json.dump(data, file, indent=2) def load_summary(path): with open(path) as file: return json.load(file)

"""Log levels as an enum: a fixed, ordered set of values, where a typo fails loudly.""" from enum import IntEnum from typing import Self # Other spellings some services use. (Not inside the class: every plain # attribute of an Enum class becomes a member.) ALIASES = {"WARN": "WARNING", "ERR": "ERROR", "FATAL": "CRITICAL"} class Level(IntEnum): """Ordered by severity. The numbers match the standard logging module.""" DEBUG = 10 INFO = 20 WARNING = 30 ERROR = 40 CRITICAL = 50 def __str__(self) -> str: return self.name @classmethod def from_text(cls, text: str) -> Self: """Level.from_text("warn") is Level.WARNING. Anything unknown raises ValueError.""" name = text.strip().upper() try: return cls[ALIASES.get(name, name)] except KeyError: raise ValueError(f"unknown level {text!r}") from None

import unittest from datetime import datetime from analyzer import ParseError, count, parse_fields, parse_lines, parse_record, summarize from levels import Level GOOD = "2026-09-01 09:00:00 INFO a: x=1" class TestParseRecord(unittest.TestCase): def test_prefix_becomes_fields(self): record = parse_record("2026-09-01 09:14:02 INFO auth: user=ada action=login") self.assertEqual(record.timestamp, datetime(2026, 9, 1, 9, 14, 2)) self.assertEqual((record.level, record.source), (Level.INFO, "auth")) def test_other_spellings_of_a_level_are_accepted(self): self.assertIs(parse_record("2026-09-01 09:14:02 WARN billing: x=1").level, Level.WARNING) def test_message_has_no_leading_space(self): self.assertEqual(parse_record("2026-09-01 09:14:02 INFO auth: user=ada").message, "user=ada") def test_each_kind_of_bad_line_is_reported(self): cases = { "garbage": "too few parts", "2026-13-01 09:00:00 INFO a: x=1": "bad timestamp", "2026-09-01 09:00:00 LOUD a: x=1": "unknown level 'LOUD'", "2026-09-01 09:00:00 INFO no colon here": "missing 'source:'", } for line, reason in cases.items(): with self.subTest(line=line): with self.assertRaises(ParseError) as caught: parse_record(line, 7) self.assertEqual(caught.exception.reason, reason) class TestParseFields(unittest.TestCase): def test_two_quoted_values_stay_separate(self): fields = parse_fields('query="python books" msg="index offline"') self.assertEqual(fields, {"query": "python books", "msg": "index offline"}) class TestParseLines(unittest.TestCase): def test_bad_lines_go_to_on_error(self): errors = [] records = list(parse_lines(["garbage", "", GOOD], on_error=errors.append)) self.assertEqual((len(records), [e.line_number for e in errors]), (1, [1])) def test_without_on_error_the_first_bad_line_stops_parsing(self): with self.assertRaises(ParseError): list(parse_lines([GOOD, "garbage", GOOD])) def test_errors_are_known_once_the_records_are_read(self): errors = [] records = parse_lines(["garbage", GOOD], on_error=errors.append) self.assertEqual(errors, []) list(records) self.assertEqual(len(errors), 1) class TestCounting(unittest.TestCase): def test_summary_works_on_a_generator(self): lines = ["2026-09-01 09:00:00 ERROR db: x=1", "2026-09-01 09:00:01 INFO db: x=2"] levels, errors = summarize(parse_lines(iter(lines))) self.assertEqual(levels, {Level.ERROR: 1, Level.INFO: 1}) self.assertEqual(errors, {"db": 1}) def test_critical_counts_as_an_error_source(self): levels, errors = summarize(parse_lines(["2026-09-01 09:00:00 CRITICAL db: x=1"])) self.assertEqual(errors, {"db": 1}) def test_count_with_a_key_and_a_filter(self): records = list(parse_lines([ "2026-09-01 09:00:00 ERROR db: x=1", "2026-09-01 10:00:00 ERROR db: x=2", "2026-09-01 10:30:00 INFO db: x=3", ])) by_hour = count(records, key=lambda r: r.timestamp.hour, keep=lambda r: r.level is Level.ERROR) self.assertEqual(by_hour, {9: 1, 10: 1})

import unittest from datetime import datetime from analyzer import parse_record from filters import all_of, between, level_at_least, source_is from levels import Level def record(level="INFO", source="auth", hour=9): return parse_record(f"2026-09-01 {hour:02}:00:00 {level} {source}: x=1") class TestFilters(unittest.TestCase): def test_level_at_least_keeps_every_worse_level(self): # Day 5 bug hunt keep = level_at_least("WARNING") self.assertEqual([keep(record(level.name)) for level in Level], [False, False, True, True, True]) def test_level_at_least_takes_a_level_or_its_name(self): self.assertTrue(level_at_least(Level.ERROR)(record("CRITICAL"))) self.assertTrue(level_at_least("warn")(record("WARNING"))) def test_an_unknown_level_fails_when_the_filter_is_made(self): with self.assertRaises(ValueError): level_at_least("WARNIGN") def test_source_is_any_of(self): keep = source_is("auth", "db") self.assertEqual([keep(record(source=s)) for s in ("auth", "db", "search")], [True, True, False]) def test_between_includes_start_excludes_end(self): keep = between(datetime(2026, 9, 1, 9), datetime(2026, 9, 1, 10)) self.assertEqual([keep(record(hour=h)) for h in (8, 9, 10)], [False, True, False]) def test_all_of(self): keep = all_of(level_at_least("ERROR"), source_is("db")) self.assertEqual([keep(record("ERROR", "db")), keep(record("ERROR", "auth")), keep(record("INFO", "db"))], [True, False, False]) def test_filters_made_in_a_loop_are_all_different(self): by_level = {level: level_at_least(level) for level in Level} self.assertTrue(by_level[Level.DEBUG](record("INFO"))) self.assertFalse(by_level[Level.ERROR](record("INFO")))

import sqlite3 import unittest from analyzer import parse_record from store import (connect, count_by, load_records, load_summary, open_store, save_records, save_summary, search, summary) LINES = [ "2026-09-01 09:14:02 INFO auth: user=ada action=login", "2026-09-01 09:15:30 ERROR payments: order=1043 msg=\"card declined\"", "2026-09-01 10:02:11 ERROR db: msg=\"can't reach replica\"", ] def records(): return [parse_record(line, number) for number, line in enumerate(LINES, start=1)] class TestStore(unittest.TestCase): def setUp(self): self.conn = connect() # a new, empty database for every test self.addCleanup(self.conn.close) def test_records_come_back_equal(self): # bug hunt 1 save_records(self.conn, "a.log", records()) self.assertEqual(load_records(self.conn, "a.log"), records()) def test_search_text_with_a_quote(self): # bug hunt 2 save_records(self.conn, "a.log", records()) found = search(self.conn, "can't") self.assertEqual([r.source for r in found], ["db"]) def test_search_is_not_sql(self): save_records(self.conn, "a.log", records()) self.assertEqual(search(self.conn, "x' OR '1'='1"), []) def test_saving_the_same_log_twice_stores_it_once(self): # bug hunt 3 self.assertEqual(save_records(self.conn, "a.log", records()), 3) self.assertEqual(save_records(self.conn, "a.log", records()), 0) self.assertEqual(count_by(self.conn, "level"), {"ERROR": 2, "INFO": 1}) def test_count_by_one_level(self): save_records(self.conn, "a.log", records()) self.assertEqual(count_by(self.conn, "source", level="ERROR"), {"db": 1, "payments": 1}) def test_count_by_refuses_other_columns(self): with self.assertRaises(ValueError): count_by(self.conn, "message; DROP TABLE records") class TestSummary(unittest.TestCase): def test_summary_survives_a_round_trip_through_json(self): # bug hunt 4 conn = connect() self.addCleanup(conn.close) save_records(conn, "a.log", records()) data = summary(conn) save_summary(data, "test-summary.json") self.assertEqual(load_summary("test-summary.json"), data) self.assertEqual(data["errors_by_hour"], [{"hour": 9, "errors": 1}, {"hour": 10, "errors": 1}]) class TestOpenStore(unittest.TestCase): def test_closes_the_connection_even_when_the_block_fails(self): # Day 4, bug hunt 3 with self.assertRaises(ZeroDivisionError): with open_store() as conn: 1 / 0 with self.assertRaises(sqlite3.ProgrammingError): # "Cannot operate on a closed database" conn.execute("SELECT 1")

import unittest from levels import Level class TestLevel(unittest.TestCase): def test_levels_are_ordered_by_severity(self): self.assertLess(Level.WARNING, Level.ERROR) self.assertEqual(sorted([Level.ERROR, Level.DEBUG, Level.CRITICAL]), [Level.DEBUG, Level.ERROR, Level.CRITICAL]) def test_from_text_accepts_other_spellings(self): for text, level in {"warn": Level.WARNING, "Error": Level.ERROR, " ERR ": Level.ERROR, "FATAL": Level.CRITICAL}.items(): with self.subTest(text=text): self.assertIs(Level.from_text(text), level) def test_from_text_rejects_anything_else(self): with self.assertRaises(ValueError): Level.from_text("LOUD") def test_prints_as_its_name(self): self.assertEqual(f"{Level.ERROR} {Level.INFO!s:>6}", "ERROR INFO")

import sys import unittest import test_analyzer import test_filters import test_levels import test_store loader = unittest.defaultTestLoader suite = unittest.TestSuite(loader.loadTestsFromModule(module) for module in (test_levels, test_analyzer, test_filters, test_store)) unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)

Hint 1: the idea

This lesson’s rule, applied at both edges. Text becomes a Level once, where a line is parsed, and everything after that compares Levels, never strings. At the other edge, the database, a Level has to become something a column can hold — and has to come back as a Level when it’s read.

Hint 2: a first step

Steps 1 and 2 are small: Level.from_text in parse_record, and comparisons of Levels in level_at_least. For step 3, print what the database really stored for one saved record, and compare it with what the failing test expected to get back. An IntEnum is an int — that’s the clue.

Show one solution

analyzer.py: convert at the edge, and give the error the same reason as before, so callers see no difference:

from levels import Level

...
    date, time, level_text, rest = parts
    ...
    try:
        level = Level.from_text(level_text)
    except ValueError:
        raise ParseError(line_number, f"unknown level {level_text!r}", line) from None

In summarize, record.level == "ERROR" compared a Level with a string. An IntEnum is never equal to a string, so it was quietly False for every record, another silent miscount. The test that caught it checks something the old code never did: CRITICAL counts as an error too.

        if record.level >= Level.ERROR:
            error_sources[record.source] += 1

filters.py: one conversion when the filter is built, and a plain comparison per record:

def level_at_least(minimum):
    """Keep records at minimum or more severe: level_at_least("WARNING")."""
    floor = Level.from_text(minimum) if isinstance(minimum, str) else minimum

    def keep(record):
        return record.level >= floor

    return keep

store.py: this is the step the tests had to find for you. After step 1, save_records passed Level members straight to SQLite. An IntEnum is an int, so SQLite stored 40, and the TEXT column turned it into the text '40'. Nothing failed. count_by(level="ERROR") found nothing, and the round trip came back with level='20'. It’s Day 3’s lesson again: convert at the storage edge, explicitly, in both directions.

from levels import Level

...
    rows = ((log, r.line_number, r.timestamp.isoformat(sep=" "), r.level.name, r.source, r.message)
            for r in records)
...
        level=Level[row["level"]],

Storing the name rather than the number keeps the database readable, and keeps it working if the numbers ever change. Level[...] on the way out raises KeyError for anything that isn’t a level, so a corrupted row fails loudly instead of turning into a string.

The order of that exercise was deliberate. Fix the code the tests point at, run them again, and let the next failure show you the next place. Nobody lists every place a value flows through a program from memory. The tests do it for you, which is the whole reason they were written on Day 1.

The Day 5 reference solution

The analyzer at the end of Day 5, and where Day 7 picks it up. (Day 6 is a standalone exercise.) Compared with the checkpoint, it also has:

report.py, with CountSection, ListSection and Report from lesson 1, typed, and with the Section protocol.
Type hints on every public function in analyzer.py, filters.py, levels.py and report.py. mypy --strict finds no problems in those four files. store.py isn’t hinted yet, and that’s a fine place to practise.
store.count_by accepts a Level or a name, and runs it through Level.from_text, so passing Level.ERROR can’t silently match nothing. Its at_least= option counts a level and everything worse.
One meaning of “errors”. summarize counts ERROR and CRITICAL as errors, so the SQL summary in store.py does too. The stored names it asks for come from Level’s order (level_names(Level.ERROR)), never from comparing the text, which would be today’s bug again. A test with one record at every level checks that Python and SQL agree.
Day 3’s guarantees, still tested. A failed save, including a stream that fails part-way, stores nothing. Search results come back oldest first, and % and _ in a search are plain text.

main.py builds the daily report from sections, then runs all thirty-eight tests.

Day 5 reference: sections, hints and levels Python

"""The Day 5 report: made of sections, with levels that can't be misspelled.""" import sys import unittest import test_analyzer import test_filters import test_levels import test_report import test_store from analyzer import ParseError, parse_lines, read_lines from filters import level_at_least from report import CountSection, ListSection, Report errors: list[ParseError] = [] daily = Report("Daily report: messy.log", [ CountSection("Levels", key=lambda r: r.level), CountSection("Warnings and worse, by source", key=lambda r: r.source, keep=level_at_least("WARNING")), ListSection("First errors", keep=level_at_least("ERROR"), limit=3), ]) print(daily.render(parse_lines(read_lines("messy.log"), on_error=errors.append))) print(f"\n({len(errors)} bad line(s) skipped)\n") loader = unittest.defaultTestLoader suite = unittest.TestSuite( loader.loadTestsFromModule(module) for module in (test_levels, test_analyzer, test_filters, test_report, test_store)) unittest.TextTestRunner(stream=sys.stdout, verbosity=1).run(suite)

"""Log analyzer, Day 5: typed records, parsed once and counted any way you like.""" import re from collections import Counter from collections.abc import Callable, Hashable, Iterable, Iterator from dataclasses import dataclass, field from datetime import datetime from levels import Level # key=value, where a value is either "quoted, spaces allowed" or a single word. FIELD = re.compile(r'(?P<key>\w+)=(?P<value>"[^"]*"|\S+)') @dataclass(frozen=True) class LogRecord: timestamp: datetime level: Level source: str message: str fields: dict[str, str] = field(default_factory=dict) line_number: int = 0 type Predicate = Callable[[LogRecord], bool] class ParseError(ValueError): """A line that is not a valid log record.""" def __init__(self, line_number: int, reason: str, line: str) -> None: super().__init__(f"line {line_number}: {reason}: {line!r}") self.line_number = line_number self.reason = reason self.line = line def parse_fields(text: str) -> dict[str, str]: """Turn 'a=1 b="two words"' into {'a': '1', 'b': 'two words'}.""" return {m["key"]: m["value"].strip('"') for m in FIELD.finditer(text)} def parse_record(line: str, line_number: int = 0) -> LogRecord: """Parse one line, or raise ParseError saying what is wrong with it.""" parts = line.split(maxsplit=3) if len(parts) < 4: raise ParseError(line_number, "too few parts", line) date, time, level_text, rest = parts try: timestamp = datetime.fromisoformat(f"{date} {time}") except ValueError: raise ParseError(line_number, "bad timestamp", line) from None try: level = Level.from_text(level_text) except ValueError: raise ParseError(line_number, f"unknown level {level_text!r}", line) from None source, colon, message = rest.partition(":") if not colon: raise ParseError(line_number, "missing 'source:'", line) message = message.strip() return LogRecord(timestamp, level, source, message, parse_fields(message), line_number) def read_lines(path: str) -> Iterator[str]: """Yield the lines of a file one at a time, without line endings.""" with open(path) as file: for line in file: yield line.rstrip("\n") def parse_lines( lines: Iterable[str], on_error: Callable[[ParseError], object] | None = None ) -> Iterator[LogRecord]: """Yield a record per good line. A bad line's ParseError is passed to on_error. With no on_error it is raised, so by default the first bad line stops everything: nothing is dropped silently. """ for number, line in enumerate(lines, start=1): if not line.strip(): continue try: record = parse_record(line, number) except ParseError as error: if on_error is None: raise on_error(error) else: yield record def count[K: Hashable]( records: Iterable[LogRecord], key: Callable[[LogRecord], K], keep: Predicate | None = None ) -> Counter[K]: """Count records by key(record), counting only those where keep(record) is true.""" return Counter(key(r) for r in records if keep is None or keep(r)) def summarize(records: Iterable[LogRecord]) -> tuple[Counter[Level], Counter[str]]: """Count levels and errors per source in a single pass.""" levels: Counter[Level] = Counter() error_sources: Counter[str] = Counter() for record in records: levels[record.level] += 1 if record.level >= Level.ERROR: error_sources[record.source] += 1 return levels, error_sources

"""Ways to choose records: each function builds a Predicate, a yes/no question about one record.""" from datetime import datetime from analyzer import LogRecord, Predicate from levels import Level def level_at_least(minimum: Level | str) -> Predicate: """Keep records at minimum or more severe: level_at_least("WARNING").""" floor = Level.from_text(minimum) if isinstance(minimum, str) else minimum def keep(record: LogRecord) -> bool: return record.level >= floor return keep def source_is(*sources: str) -> Predicate: """Keep records from any of the given sources.""" wanted = set(sources) def keep(record: LogRecord) -> bool: return record.source in wanted return keep def between(start: datetime, end: datetime) -> Predicate: """Keep records from start (included) up to end (not included).""" def keep(record: LogRecord) -> bool: return start <= record.timestamp < end return keep def all_of(*predicates: Predicate) -> Predicate: """Keep records that every predicate keeps.""" def keep(record: LogRecord) -> bool: return all(predicate(record) for predicate in predicates) return keep

"""Reports made of sections. Each section is a small object that renders itself.""" from collections.abc import Callable, Hashable, Iterable, Sequence from dataclasses import dataclass from typing import Protocol from analyzer import LogRecord, Predicate, count class Section(Protocol): """Anything with a title and a render method can be part of a Report.""" @property def title(self) -> str: ... # read-only, so frozen dataclasses qualify def render(self, records: Sequence[LogRecord]) -> list[str]: ... @dataclass(frozen=True) class CountSection: """Records counted by key, busiest first.""" title: str key: Callable[[LogRecord], Hashable] keep: Predicate | None = None top: int | None = None def render(self, records: Sequence[LogRecord]) -> list[str]: counts = count(records, self.key, self.keep).most_common(self.top) return [f" {str(name):<12}{n:>4}" for name, n in counts] or [" (none)"] @dataclass(frozen=True) class ListSection: """The first few records a predicate keeps, one per line.""" title: str keep: Predicate limit: int = 5 def render(self, records: Sequence[LogRecord]) -> list[str]: chosen = [r for r in records if self.keep(r)][: self.limit] return [f" {r.timestamp:%H:%M:%S} {r.level!s:<8} {r.source}: {r.message}" for r in chosen] or [" (none)"] @dataclass(frozen=True) class Report: title: str sections: Sequence[Section] def render(self, records: Iterable[LogRecord]) -> str: rows = list(records) # every section reads them, so read the stream once lines = [self.title, "=" * len(self.title)] for section in self.sections: lines += ["", section.title, *section.render(rows)] return "\n".join(lines)

"""Log analyzer, Day 5: keep parsed records in SQLite, and ask questions in SQL.""" import json import sqlite3 from contextlib import contextmanager from datetime import datetime from analyzer import LogRecord, parse_fields from levels import Level SCHEMA = """ CREATE TABLE IF NOT EXISTS records ( log TEXT NOT NULL, line_number INTEGER NOT NULL, timestamp TEXT NOT NULL, -- ISO 8601, so text order is time order level TEXT NOT NULL, source TEXT NOT NULL, message TEXT NOT NULL, PRIMARY KEY (log, line_number) ) """ def connect(path=":memory:"): """Open (or create) a database with the records table in it.""" conn = sqlite3.connect(path) conn.row_factory = sqlite3.Row conn.execute(SCHEMA) return conn @contextmanager def open_store(path=":memory:"): """Connect, and close the connection when the with block ends, however it ends.""" conn = connect(path) try: yield conn finally: conn.close() def save_records(conn, log, records): """Store records read from one log. Lines already stored are skipped. Returns how many new records were stored. """ rows = ((log, r.line_number, r.timestamp.isoformat(sep=" "), r.level.name, r.source, r.message) for r in records) with conn: # one transaction: all rows, or none cursor = conn.executemany( "INSERT INTO records VALUES (?, ?, ?, ?, ?, ?) " "ON CONFLICT (log, line_number) DO NOTHING", rows) return cursor.rowcount def record_from_row(row): """Rebuild a LogRecord. Fields are re-parsed from the message, not stored twice.""" return LogRecord( timestamp=datetime.fromisoformat(row["timestamp"]), level=Level[row["level"]], source=row["source"], message=row["message"], fields=parse_fields(row["message"]), line_number=row["line_number"], ) def load_records(conn, log): rows = conn.execute( "SELECT * FROM records WHERE log = ? ORDER BY line_number", (log,)) return [record_from_row(row) for row in rows] def like_pattern(text): """A LIKE pattern matching text anywhere, with LIKE's own wildcards made literal.""" for special in ("\\", "%", "_"): # the escape character goes first text = text.replace(special, "\\" + special) return f"%{text}%" def search(conn, text): """Records whose message contains text (ignoring ASCII case), oldest first.""" rows = conn.execute( "SELECT * FROM records WHERE message LIKE ? ESCAPE '\\' ORDER BY timestamp", (like_pattern(text),)) return [record_from_row(row) for row in rows] COUNTABLE = {"level", "source"} def level_names(at_least): """The stored names of every level this severe or worse, chosen by Level's order.""" floor = Level.from_text(str(at_least)) return tuple(level.name for level in Level if level >= floor) def count_by(conn, column, level=None, *, at_least=None): """Count records per level or per source, busiest first. level counts one level only; at_least counts that level and every worse one. """ if column not in COUNTABLE: # a column name can't be a ? parameter raise ValueError(f"can't count by {column!r}; choose from {sorted(COUNTABLE)}") # Level.from_text accepts "ERROR" or Level.ERROR, and rejects typos. names = () if level is not None: names = (Level.from_text(str(level)).name,) elif at_least is not None: names = level_names(at_least) where = f"WHERE level IN ({', '.join('?' * len(names))})" if names else "" rows = conn.execute( f"SELECT {column} AS key, COUNT(*) AS n FROM records {where} " f"GROUP BY {column} ORDER BY n DESC, key", names) return {row["key"]: row["n"] for row in rows} def summary(conn): """Everything the daily report needs, as plain JSON-ready values.""" total, first, last = conn.execute( "SELECT COUNT(*), MIN(timestamp), MAX(timestamp) FROM records").fetchone() errors = level_names(Level.ERROR) # "errors" means ERROR or worse, everywhere hours = conn.execute( "SELECT CAST(substr(timestamp, 12, 2) AS INTEGER) AS hour, COUNT(*) AS n " f"FROM records WHERE level IN ({', '.join('?' * len(errors))}) " "GROUP BY hour ORDER BY hour", errors) return { "records": total, "first": first, "last": last, "levels": count_by(conn, "level"), "errors_by_source": count_by(conn, "source", at_least=Level.ERROR), "errors_by_hour": [{"hour": row["hour"], "errors": row["n"]} for row in hours], } def save_summary(data, path): with open(path, "w") as file: json.dump(data, file, indent=2) def load_summary(path): with open(path) as file: return json.load(file)

import unittest from analyzer import parse_lines from filters import level_at_least from report import CountSection, ListSection, Report LINES = [ "2026-09-01 09:14:02 INFO auth: user=ada action=login", "2026-09-01 09:15:30 ERROR payments: order=1043", "2026-09-01 09:17:45 WARNING payments: order=1044 retry=2", "2026-09-01 10:02:11 ERROR payments: order=1050", ] class TestReport(unittest.TestCase): def test_count_section_is_busiest_first(self): section = CountSection("By source", key=lambda r: r.source, top=1) self.assertEqual(section.render(list(parse_lines(LINES))), [" payments 3"]) def test_empty_sections_say_so(self): section = ListSection("Critical", keep=level_at_least("CRITICAL")) self.assertEqual(section.render(list(parse_lines(LINES))), [" (none)"]) def test_every_section_sees_every_record(self): # Day 5, bug hunt 1 report = Report("Test", [CountSection("A", key=lambda r: r.level), CountSection("B", key=lambda r: r.level)]) text = report.render(parse_lines(LINES)) # a generator, read once self.assertEqual(text.count("INFO 1"), 2)

import sqlite3 import unittest from dataclasses import replace from analyzer import ParseError, parse_lines, parse_record, summarize from levels import Level from store import (connect, count_by, load_records, load_summary, open_store, save_records, save_summary, search, summary) LINES = [ "2026-09-01 09:14:02 INFO auth: user=ada action=login", "2026-09-01 09:15:30 ERROR payments: order=1043 msg=\"card declined\"", "2026-09-01 10:02:11 ERROR db: msg=\"can't reach replica\"", ] def records(): return [parse_record(line, number) for number, line in enumerate(LINES, start=1)] class TestStore(unittest.TestCase): def setUp(self): self.conn = connect() # a new, empty database for every test self.addCleanup(self.conn.close) def test_records_come_back_equal(self): # bug hunt 1 save_records(self.conn, "a.log", records()) self.assertEqual(load_records(self.conn, "a.log"), records()) def test_search_text_with_a_quote(self): # bug hunt 2 save_records(self.conn, "a.log", records()) found = search(self.conn, "can't") self.assertEqual([r.source for r in found], ["db"]) def test_search_is_not_sql(self): save_records(self.conn, "a.log", records()) self.assertEqual(search(self.conn, "x' OR '1'='1"), []) def test_saving_the_same_log_twice_stores_it_once(self): # bug hunt 3 self.assertEqual(save_records(self.conn, "a.log", records()), 3) self.assertEqual(save_records(self.conn, "a.log", records()), 0) self.assertEqual(count_by(self.conn, "level"), {"ERROR": 2, "INFO": 1}) def test_search_results_are_oldest_first(self): # Day 3 exercise save_records(self.conn, "a.log", reversed(records())) # stored newest first self.assertEqual([r.source for r in search(self.conn, "msg=")], ["payments", "db"]) def test_search_treats_wildcards_as_text(self): save_records(self.conn, "a.log", records()) self.assertEqual(search(self.conn, "%"), []) self.assertEqual(search(self.conn, "_"), []) def test_a_failed_save_stores_nothing(self): # Day 3 exercise good = records() bad = replace(good[1], source=None) # NOT NULL: the database refuses it with self.assertRaises(sqlite3.IntegrityError): save_records(self.conn, "a.log", [good[0], bad, good[2]]) self.assertEqual(count_by(self.conn, "level"), {}) def test_a_stream_that_fails_part_way_stores_nothing(self): def stream(): yield records()[0] raise ParseError(2, "too few parts", "garbage") # fail-fast parsing, mid-save with self.assertRaises(ParseError): save_records(self.conn, "a.log", stream()) self.assertEqual(count_by(self.conn, "level"), {}) def test_count_by_one_level(self): save_records(self.conn, "a.log", records()) self.assertEqual(count_by(self.conn, "source", level="ERROR"), {"db": 1, "payments": 1}) def test_count_by_refuses_other_columns(self): with self.assertRaises(ValueError): count_by(self.conn, "message; DROP TABLE records") class TestSummary(unittest.TestCase): def test_summary_survives_a_round_trip_through_json(self): # bug hunt 4 conn = connect() self.addCleanup(conn.close) save_records(conn, "a.log", records()) data = summary(conn) save_summary(data, "test-summary.json") self.assertEqual(load_summary("test-summary.json"), data) self.assertEqual(data["errors_by_hour"], [{"hour": 9, "errors": 1}, {"hour": 10, "errors": 1}]) class TestOpenStore(unittest.TestCase): def test_closes_the_connection_even_when_the_block_fails(self): # Day 4, bug hunt 3 with self.assertRaises(ZeroDivisionError): with open_store() as conn: 1 / 0 with self.assertRaises(sqlite3.ProgrammingError): # "Cannot operate on a closed database" conn.execute("SELECT 1") class TestErrorsMeanTheSameEverywhere(unittest.TestCase): def test_error_or_worse_in_python_and_in_sql(self): lines = [f"2026-09-01 09:0{i}:00 {level.name} db: x={i}" for i, level in enumerate(Level)] records = list(parse_lines(lines)) conn = connect() self.addCleanup(conn.close) save_records(conn, "all.log", records) _, error_sources = summarize(records) self.assertEqual(error_sources, {"db": 2}) # ERROR and CRITICAL self.assertEqual(summary(conn)["errors_by_source"], {"db": 2}) self.assertEqual(summary(conn)["errors_by_hour"], [{"hour": 9, "errors": 2}]) self.assertEqual(count_by(conn, "source", at_least="warning"), {"db": 3})

2026-09-01 09:14:02 INFO auth: user=ada action=login 2026-09-01 09:14:07 WARNING payments: order=1042 amount=19.99 retry=1 2026-09-01 09:15:30 ERROR payments: order=1043 msg="card declined" 2026-09-01 09:16:11 INFO auth: user=alan action=logout --- rotated at 09:16 --- 2026-09-01 09:17:45 ERROR search: query="python books" msg="index offline" 2026-09-01 9:18 INFO search: query=dune results=4 2026-09-01 09:19:30 ERROR auth: user=grace msg="token expired"

Working on your own machine? Download the Day 5 files and carry on there. You need Python 3.12 or later, and nothing else.

What you can now do

Carry a change of type through a codebase in small, test-guided steps.
Recognise the places a value crosses a boundary (parsing, storage, comparison) and convert it at each one.
Spot the silent failures a type change can cause, like an IntEnum compared with a string or stored as a number.

That’s the end of the analyzer’s core. Tomorrow is a standalone day about doing several slow things at once. The analyzer comes back on Day 7, to be finished and shipped.

Day 5 checkpoint

The Day 5 reference solution

What you can now do

Was this article helpful?