Edge Cases, on Purpose

The Course
•
September 18, 2026
•
9 min read

Warm-up: from Day 6

The last warm-up of the course, from memory first.

1. Predict both lines. Same three jobs, two ways of waiting for them.

Warm-up: predict Python
import asyncio

async def job(name, delay):
    await asyncio.sleep(delay)
    return name

async def main():
    print(await asyncio.gather(job("a", 0.03), job("b", 0.01), job("c", 0.02)))
    order = []
    for next_done in asyncio.as_completed([job("a", 0.03), job("b", 0.01), job("c", 0.02)]):
        order.append(await next_done)
    print(order)

asyncio.run(main())
Check your prediction
['a', 'b', 'c']
['b', 'c', 'a']

gather returns results in the order you asked; as_completed hands them over in the order they finish. From Tasks and the Event Loop.

2. Fix. One source being down should be reported, not take the whole report with it. This should print ['web: ok', 'db: failed', 'api: ok'].

Warm-up: fix Python
import asyncio

async def fetch(name):
    if name == "db":
        raise ConnectionError("db unreachable")
    return f"{name}: ok"

async def main():
    print(await asyncio.gather(fetch("web"), fetch("db"), fetch("api")))

asyncio.run(main())
Check your fix
import asyncio

async def fetch(name):
    if name == "db":
        raise ConnectionError("db unreachable")
    return f"{name}: ok"

async def safely(name):
    try:
        return await fetch(name)
    except ConnectionError:
        return f"{name}: failed"

async def main():
    print(await asyncio.gather(safely("web"), safely("db"), safely("api")))

asyncio.run(main())
['web: ok', 'db: failed', 'api: ok']

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 ended 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 """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:

Edge caseHandled byTest
Blank or whitespace-only linesparse_lines skips themtest_bad_lines_go_to_on_error
Junk lines, bad times, unknown levelsParseError with a reasontest_each_kind_of_bad_line_is_reported
A bad line is never silently droppedon_error, fail-fast by defaulttest_without_on_error_the_first_bad_line_stops_parsing
Quoted values with spaces, several quotes per linethe FIELD regextest_two_quoted_values_stay_separate
A stream read twiceone-pass summarize, Report makes one listtest_summary_works_on_a_generator
Quotes in search text? placeholderstest_search_text_with_a_quote
The same log imported twiceprimary key, ON CONFLICT DO NOTHINGtest_saving_the_same_log_twice_stores_it_once
A save that fails half-wayone transactiontest_a_failed_save_stores_nothing, test_a_stream_that_fails_part_way_stores_nothing
% or _ in search textescaped LIKE patterntest_search_treats_wildcards_as_text
”Errors” in Python and in SQLlevel_names(Level.ERROR)test_error_or_worse_in_python_and_in_sql
Integer keys in JSONlists of objectstest_summary_survives_a_round_trip_through_json
WARN, err, ErrorLevel.from_texttest_from_text_accepts_other_spellings
Levels compared by spellingIntEnumtest_level_at_least_keeps_every_worse_level

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 export Python
"""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:

def test_a_marked_first_line_parses(self):
    lines = self.read("\ufeff2026-09-01 09:14:02 INFO auth: user=ada\n".encode())
    self.assertEqual(parse_record(lines[0]).source, "auth")

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 UnicodeDecodeError halfway 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 os
import sys
import unittest


def 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.

Last updated on September 18, 2026

Was this article helpful?

Your response is saved on this device.