Handle the failure per source, so each one turns its own expected error into
a result. Day 6’s bug hunt, one server down, no report at all, was this.
3. Explain. No code: fetch_safely catches TimeoutError and
ConnectionError — why not every Exception?
Check your explanation
Those two are the expected ways a network source fails, and the report can
say something useful about each. Anything else — a KeyError, a typo — is a
bug in the program, and catching it would disguise the bug as “that server
failed”. From Timeouts and Failures.
Where Day 5 ended
The last day. Here’s where the analyzer stood at the end of Day 5:
Where Day 5 endedPython"""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
"""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
"""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 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 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 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 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"
Today it becomes a tool other people can run: a command line, proper logging,
a README. But first, one more look at what it does with awkward input. A tool
used by other people meets inputs its author never imagined, so the edge cases
are worth finding on purpose, before they find you.
An inventory of the week
Most edge cases are already handled, because every bug hunt this week left a
regression test behind. Listing them shows what’s covered, and where the gaps
are:
What’s missing is the one edge nobody has looked at yet: the file itself. Every
test so far starts from lines that are already strings. The bytes on disk, and
how they become text, have never been tested.
Bug hunt: the first line is always bad
The billing team exports its logs with a Windows tool. Their file looks exactly
like messy.log in any editor, but the report says one line more is bad:
The billing exportPython"""The nightly report, now reading an export from the billing team's tool."""
from analyzer import parse_lines, read_lines
# The billing tool saves its exports like this. The file looks exactly like
# messy.log in any editor.
with open("messy.log") as original, open("export.log", "w", encoding="utf-8-sig") as export:
export.write(original.read())
errors = []
records = list(parse_lines(read_lines("export.log"), on_error=errors.append))
print(f"{len(records)} records, {len(errors)} bad line(s):")
for error in errors:
print(" ", error)
"""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
"""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
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"
Reproduce. Three bad lines instead of two, and the new one is line 1,
which looks perfect.
Inspect. The error message shows the line with repr, which is exactly
what you want here. Look at the start of it: '\ufeff2026-09-01 09:14:02 ....
There’s a character before the date that no editor displays.
Your turn, before you open this. Write a test that reproduces the bug from raw bytes, without export.log, then fix read_lines. Read the reveal either way: it introduces the utf-8-sig encoding the rest of the lesson builds on.
The diagnosis and the fix (read it either way)
Narrow.U+FEFF is the byte-order mark, or BOM. Some Windows tools write
it at the very start of a UTF-8 file, as the bytes EF BB BF, to say “this file
is UTF-8”. Python’s plain "utf-8" encoding reads it as an ordinary character,
so the first line starts with an invisible \ufeff, and fromisoformat rightly
refuses it. You can see the bytes for yourself:
print(open("export.log", "rb").read(12))
Pin it. A test that feeds read_lines a file with a BOM and checks the first
line parses:
Fix. Tell open which encoding the file is in. The "utf-8-sig" encoding
reads UTF-8 and skips a BOM if there is one. In analyzer.py:
with open(path, encoding="utf-8-sig", errors="replace") as file:
Run main.py again: two bad lines, as in messy.log.
Always say the encoding
open(path) without an encoding uses a default that depends on the computer.
It’s UTF-8 on most, but it isn’t on some Windows setups, so the same program can
read the same file differently on different machines. Naming the encoding makes
the program’s behaviour part of the code. (Python 3.15 makes UTF-8 the default
everywhere.)
Bytes that aren’t text
The fix above has a second argument: errors="replace". A log file that
isn’t valid UTF-8 at all, say one byte from an old Latin-1 system, makes a plain
open raise UnicodeDecodeErrorhalfway through the file, and the whole
report is lost for one byte.
That’s the Day 2 decision again, one level down. Fail on the first bad byte, or
carry on and let the damage show? errors="replace" turns each undecodable byte
into U+FFFD, the replacement character �. The line usually still parses,
the odd character shows up in the message, and one bad byte costs one character
instead of the whole file.
Your turn
Exercise
Make read_lines survive real files
read_lines below is the Day 5 version. The test feeds it six awkward files as
raw bytes: an empty file, a file with no final newline, Windows line endings, a
BOM, a byte that isn’t UTF-8, and genuine UTF-8. Two cases fail. Fix
read_lines so all six pass.
import osimport sysimport unittestdef read_lines(path): """Yield the lines of a text file one at a time, without line endings.""" with open(path) as file: for line in file: yield line.rstrip("\n")class TestReadLines(unittest.TestCase): def read(self, data): """Write data to a file as raw bytes, and read it back with read_lines.""" with open("edge-case.log", "wb") as file: file.write(data) try: return list(read_lines("edge-case.log")) finally: os.remove("edge-case.log") def test_edge_cases(self): cases = { b"": [], b"one\ntwo": ["one", "two"], # no newline at the end b"one\r\ntwo\r\n": ["one", "two"], # Windows line endings b"\xef\xbb\xbfone\n": ["one"], # a byte-order mark b"caf\xe9\n": ["caf\ufffd"], # not UTF-8: replaced, not fatal "zoë\n".encode(): ["zoë"], # UTF-8 is read as UTF-8 } for data, lines in cases.items(): with self.subTest(data=data): self.assertEqual(self.read(data), lines)suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestReadLines)unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
Hint 1: the idea
Both failures happen before your code sees a single string — between the
bytes on disk and the text Python hands you. So the fix is in how the file is
opened, not in the loop: an encoding that knows about a byte-order mark, and a
rule for bytes that aren’t valid UTF-8, so one bad byte can’t stop the whole
file.
Hint 2: a first step
Run the tests and note which two fail and with what error. The bug hunt above
introduced the encoding that skips a BOM; its closing paragraphs describe the
second argument to open that deals with the undecodable byte.
Show one solution
def read_lines(path): """Yield the lines of a text file one at a time, without line endings. A byte-order mark at the start is skipped, and bytes that aren't valid UTF-8 become U+FFFD instead of stopping the whole file. """ with open(path, encoding="utf-8-sig", errors="replace") as file: for line in file: yield line.rstrip("\n")
Four of the six cases already passed, and it’s worth knowing why:
Windows line endings are handled by text mode itself. Reading a file with
open(path) translates \r\n to \n. That’s called universal newlines,
and it’s why rstrip("\n") is enough.
No final newline works because iterating over a file yields the last line
whether or not it ends in \n.
Genuine UTF-8 passed because this runtime’s default encoding happens to
be UTF-8. On a machine where it isn’t, zoë would have come back mangled. The
explicit encoding makes this case pass everywhere.
The test builds each file from raw bytes, not text. That’s the only way to
test what happens between the disk and your first string, which is where both
of today’s bugs lived.
What you can now do
Take an inventory of a program’s edge cases, and see which have tests.
Test file handling from raw bytes, including BOMs, bad bytes and line endings.
Open text files with an explicit encoding, and choose what happens to bytes
that can’t be decoded.
Next: replacing the analyzer’s scattered prints with logging.
Your progress is saved in this browser.
Sign in
to keep it on every device.