One feature to go, then the analyzer ships. “Ships” means someone who has never
seen this course can pick it up, run it, trust its output, and change it without
breaking it. That takes three things beyond working code: tests, a command line,
and a README.
The final checkpoint
Exercise
Add --since to logan
cli.py is the full command line from the last lesson. test_cli.py has its
ten tests, all passing, plus a new TestSince class describing the feature,
all failing:
--since HH:MM counts only records from that time of day onwards.
It combines with the other filters, so --since 09:16 --level error counts
errors from 09:16 on.
A bad time is a usage error, exit code 2, with the message from the last
lesson’s exercise.
Use your time_of_day from the last lesson. The ten existing tests must still
pass when you’re done.
"""logan: summarise server logs from the command line."""
import argparse
import logging
import sys
from collections.abc import Sequence
from analyzer import ParseError, parse_lines, read_lines
from filters import all_of, level_at_least, source_is
from levels import Level
from report import CountSection, ListSection, Report
from store import load_records, open_store, save_records, save_summary, summary
logger = logging.getLogger("logan")
def level(text: str) -> Level:
"""argparse type for --level: a level's name, in any case, or an alias."""
try:
return Level.from_text(text)
except ValueError:
names = ", ".join(level.name.lower() for level in Level)
raise argparse.ArgumentTypeError(f"unknown level {text!r} (choose from {names})") from None
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="logan", description="Summarise server logs: levels, busy sources, first errors.")
parser.add_argument("logfile", nargs="+", help="log file(s) to read")
parser.add_argument("--level", type=level, default=Level.DEBUG,
help="only count records at this level or worse (default: all)")
parser.add_argument("--source", action="append", default=[], metavar="NAME",
help="only count records from this source (repeatable)")
parser.add_argument("--top", type=int, default=5, metavar="N",
help="how many sources to list (default: 5)")
parser.add_argument("--json", metavar="PATH",
help="also write a JSON summary of every record read, ignoring the filters")
parser.add_argument("--strict", action="store_true",
help="stop at the first bad line instead of skipping it")
parser.add_argument("-v", "--verbose", action="store_true", help="log progress to stderr")
return parser
def main(argv: Sequence[str] | None = None) -> int:
"""Run logan with these arguments (default: the command line). Returns the exit code."""
args = build_parser().parse_args(argv)
logging.basicConfig(level=logging.INFO if args.verbose else logging.WARNING,
format="%(levelname)s %(name)s: %(message)s", force=True)
skipped: list[ParseError] = []
def skip(error: ParseError) -> None:
logger.warning("skipped %s", error)
skipped.append(error)
keep = level_at_least(args.level)
if args.source:
keep = all_of(keep, source_is(*args.source))
paths = list(dict.fromkeys(args.logfile)) # the same file named twice counts once
with open_store() as conn:
for path in paths:
try:
save_records(conn, path, parse_lines(read_lines(path), None if args.strict else skip))
except OSError as error: # missing, a directory, no permission...
logger.error("can't read %s: %s", path, error.strerror or error)
return 1
except ParseError as error:
logger.error("%s: %s", path, error)
return 1
records = [r for path in paths for r in load_records(conn, path) if keep(r)]
if args.json:
try:
save_summary(summary(conn), args.json)
except OSError as error:
logger.error("can't write %s: %s", args.json, error.strerror or error)
return 1
logger.info("wrote %s", args.json)
title = f"logan: {', '.join(paths)}"
print(Report(title, [
CountSection("Levels", key=lambda r: r.level),
CountSection("Busiest sources", key=lambda r: r.source, top=args.top),
ListSection("First errors", keep=level_at_least(Level.ERROR), limit=3),
]).render(records))
print(f"\n{len(records)} record(s) counted, {len(skipped)} bad line(s) skipped")
return 0
if __name__ == "__main__":
sys.exit(main())
import io
import json
import logging
import os
import unittest
from contextlib import redirect_stderr, redirect_stdout
from datetime import time
import cli
from cli import main
def run(*argv):
"""Run the command line in-process: (exit code, stdout, stderr)."""
root = logging.getLogger()
saved = root.handlers[:], root.level # main() configures logging: put it back after
out, err = io.StringIO(), io.StringIO()
try:
with redirect_stdout(out), redirect_stderr(err):
code = main(list(argv))
except SystemExit as exit: # argparse exits on --help and usage errors
code = exit.code
finally:
root.handlers[:], _ = saved
root.setLevel(saved[1])
return code, out.getvalue(), err.getvalue()
class TestCommandLine(unittest.TestCase):
def test_report_goes_to_stdout_and_warnings_to_stderr(self):
code, out, err = run("messy.log")
self.assertEqual(code, 0)
self.assertIn("6 record(s) counted, 2 bad line(s) skipped", out)
self.assertIn("skipped line 5", err)
self.assertNotIn("skipped line", out)
def test_level_and_source_filters(self):
code, out, _ = run("messy.log", "--level", "warn", "--source", "payments")
self.assertIn("2 record(s) counted", out)
def test_json_summary(self):
self.addCleanup(os.remove, "summary.json")
run("messy.log", "--json", "summary.json")
with open("summary.json") as file:
self.assertEqual(json.load(file)["records"], 6)
def test_verbose_logs_progress(self):
self.assertNotIn("stored 6", run("messy.log")[2])
self.assertIn("stored 6 new record(s)", run("messy.log", "-v")[2])
def test_strict_stops_at_the_first_bad_line(self):
code, _, err = run("messy.log", "--strict")
self.assertEqual(code, 1)
self.assertIn("line 5: bad timestamp", err)
def test_a_missing_file_is_an_error(self):
code, _, err = run("no-such.log")
self.assertEqual(code, 1)
self.assertIn("can't read no-such.log: No such file", err)
def test_a_directory_is_an_error_not_a_crash(self):
os.makedirs("a-directory", exist_ok=True)
self.addCleanup(os.rmdir, "a-directory")
code, _, err = run("a-directory")
self.assertEqual(code, 1)
self.assertIn("can't read a-directory", err)
def test_the_same_file_twice_counts_once(self):
_, out, _ = run("messy.log", "messy.log")
self.assertIn("6 record(s) counted, 2 bad line(s) skipped", out)
def test_json_summarises_every_record_whatever_the_filters(self):
self.addCleanup(os.remove, "summary.json")
_, out, _ = run("messy.log", "--level", "error", "--json", "summary.json")
self.assertIn("3 record(s) counted", out)
with open("summary.json") as file:
self.assertEqual(json.load(file)["records"], 6)
def test_an_unknown_level_is_a_usage_error(self):
code, _, err = run("messy.log", "--level", "loud")
self.assertEqual(code, 2)
self.assertIn("unknown level 'loud'", err)
class TestSince(unittest.TestCase): # the Day 7 checkpoint
def test_time_of_day(self):
self.assertEqual(cli.time_of_day("09:30"), time(9, 30))
self.assertEqual(cli.time_of_day("9:05"), time(9, 5))
def test_since_counts_only_later_records(self):
code, out, _ = run("messy.log", "--since", "09:16")
self.assertEqual(code, 0)
self.assertIn("3 record(s) counted", out)
def test_since_combines_with_level(self):
_, out, _ = run("messy.log", "--since", "09:16", "--level", "error")
self.assertIn("2 record(s) counted", out)
def test_a_bad_time_is_a_usage_error(self):
for text in ["24:00", "noon", "09:30:15"]:
with self.subTest(text=text):
code, _, err = run("messy.log", "--since", text)
self.assertEqual(code, 2)
self.assertIn(f"expected a time like 09:30, not {text!r}", err)
import sys
import unittest
import test_cli
suite = unittest.defaultTestLoader.loadTestsFromModule(test_cli)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
"""Log analyzer: 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 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")
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)
"""Keep parsed records in SQLite, and ask questions in SQL."""
import json
import logging
import sqlite3
from collections.abc import Iterable, Iterator
from contextlib import contextmanager
from datetime import datetime
from typing import Any
from analyzer import LogRecord, parse_fields
from levels import Level
logger = logging.getLogger(__name__)
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: str = ":memory:") -> sqlite3.Connection:
"""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: str = ":memory:") -> Iterator[sqlite3.Connection]:
"""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: sqlite3.Connection, log: str, records: Iterable[LogRecord]) -> int:
"""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)
logger.info("%s: stored %d new record(s)", log, cursor.rowcount)
return cursor.rowcount
def record_from_row(row: sqlite3.Row) -> LogRecord:
"""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: sqlite3.Connection, log: str) -> list[LogRecord]:
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: sqlite3.Connection, text: str) -> list[LogRecord]:
"""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: sqlite3.Connection, column: str, level: Level | str | None = None
) -> dict[str, int]:
"""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)}")
# Level.from_text accepts "ERROR" or Level.ERROR, and rejects typos.
where, params = ("WHERE level = ?", (Level.from_text(str(level)).name,)) 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: sqlite3.Connection) -> dict[str, Any]:
"""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: dict[str, Any], path: str) -> None:
with open(path, "w") as file:
json.dump(data, file, indent=2)
def load_summary(path: str) -> Any:
with open(path) as file:
return json.load(file)
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"
Hint 1: the idea
A new filter is just another keep-function, and cli.py already knows how to
combine keep-functions — that’s Day 4. So --since needs three small things:
an argument whose type is your time_of_day (argparse then turns a bad time
into a usage error by itself), a filter that keeps records at or after that time
of day, and one more line where the existing filters are joined together.
Hint 2: a first step
Add --since next to --level, with type=time_of_day, and run: the bad-time
test should now pass without any more code. Then find where keep is built —
the --source handling shows the pattern — and do the same when args.since is
set. A record’s time of day is record.timestamp.time().
Show one solution
Three additions to cli.py. The type goes above build_parser:
from datetime import datetime, timedef time_of_day(text: str) -> time: """argparse type for --since: "09:30" becomes time(9, 30).""" try: return datetime.strptime(text, "%H:%M").time() except ValueError: raise argparse.ArgumentTypeError(f"expected a time like 09:30, not {text!r}") from None
The option goes in build_parser:
parser.add_argument("--since", type=time_of_day, metavar="HH:MM", help="only count records from this time of day onwards")
And one more filter in main, built the same way as --source:
if args.since: since = args.since keep = all_of(keep, lambda record: record.timestamp.time() >= since)
Nothing outside cli.py changed. That’s Day 4’s design paying off: the
command line composes a filter from small parts, and adding an option means
adding a part. since = args.since gives the lambda its own name to close
over, so it doesn’t depend on the args object staying the same.
A between(start, end) filter already exists in filters.py, but it compares
full datetimes. --since asks about the time of day, whatever the date, so a
new filter is the honest choice. If a second option ever needs it, it can move
into filters.py as time_at_least(t).
The README
Code tells a reader how. A README tells them what, why and how to start,
before they’ve read any code. The analyzer’s README, in the reference below,
has six short sections, and each answers a question a newcomer will ask:
Section
The question it answers
One-paragraph summary and an example run
What is this, and what does it look like?
Log format
Will it read my files?
Usage
What can I type?
Exit codes
Can I use it in a script?
Design
Where do I look to change something?
Development
How do I run the tests and the type checker?
Two habits make a README worth trusting. Show real output, pasted from a
real run, not made up. And keep it next to the code it describes, so a change
to the command line and a change to its documentation land together.
The finished analyzer
This is where the week ends: six modules, six test files, a README, and a
main.py that runs logan the way a user would, then runs all fifty-four tests.
Day 7 reference: loganPython"""Day 7: run logan the way a user would, then run every test."""
import sys
import unittest
import test_analyzer
import test_cli
import test_filters
import test_levels
import test_report
import test_store
from cli import main
print("$ logan messy.log --level warning\n")
exit_code = main(["messy.log", "--level", "warning"])
print(f"\n(exit code {exit_code})\n")
loader = unittest.defaultTestLoader
suite = unittest.TestSuite(
loader.loadTestsFromModule(module)
for module in (test_levels, test_analyzer, test_filters, test_report, test_store, test_cli))
unittest.TextTestRunner(stream=sys.stdout, verbosity=1).run(suite)
"""logan: summarise server logs from the command line."""
import argparse
import logging
import sys
from collections.abc import Sequence
from datetime import datetime, time
from analyzer import ParseError, parse_lines, read_lines
from filters import all_of, level_at_least, source_is
from levels import Level
from report import CountSection, ListSection, Report
from store import load_records, open_store, save_records, save_summary, summary
logger = logging.getLogger("logan")
def level(text: str) -> Level:
"""argparse type for --level: a level's name, in any case, or an alias."""
try:
return Level.from_text(text)
except ValueError:
names = ", ".join(level.name.lower() for level in Level)
raise argparse.ArgumentTypeError(f"unknown level {text!r} (choose from {names})") from None
def time_of_day(text: str) -> time:
"""argparse type for --since: "09:30" becomes time(9, 30)."""
try:
return datetime.strptime(text, "%H:%M").time()
except ValueError:
raise argparse.ArgumentTypeError(f"expected a time like 09:30, not {text!r}") from None
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="logan", description="Summarise server logs: levels, busy sources, first errors.")
parser.add_argument("logfile", nargs="+", help="log file(s) to read")
parser.add_argument("--level", type=level, default=Level.DEBUG,
help="only count records at this level or worse (default: all)")
parser.add_argument("--source", action="append", default=[], metavar="NAME",
help="only count records from this source (repeatable)")
parser.add_argument("--since", type=time_of_day, metavar="HH:MM",
help="only count records from this time of day onwards")
parser.add_argument("--top", type=int, default=5, metavar="N",
help="how many sources to list (default: 5)")
parser.add_argument("--json", metavar="PATH",
help="also write a JSON summary of every record read, ignoring the filters")
parser.add_argument("--strict", action="store_true",
help="stop at the first bad line instead of skipping it")
parser.add_argument("-v", "--verbose", action="store_true", help="log progress to stderr")
return parser
def main(argv: Sequence[str] | None = None) -> int:
"""Run logan with these arguments (default: the command line). Returns the exit code."""
args = build_parser().parse_args(argv)
logging.basicConfig(level=logging.INFO if args.verbose else logging.WARNING,
format="%(levelname)s %(name)s: %(message)s", force=True)
skipped: list[ParseError] = []
def skip(error: ParseError) -> None:
logger.warning("skipped %s", error)
skipped.append(error)
keep = level_at_least(args.level)
if args.source:
keep = all_of(keep, source_is(*args.source))
if args.since:
since = args.since
keep = all_of(keep, lambda record: record.timestamp.time() >= since)
paths = list(dict.fromkeys(args.logfile)) # the same file named twice counts once
with open_store() as conn:
for path in paths:
try:
save_records(conn, path, parse_lines(read_lines(path), None if args.strict else skip))
except OSError as error: # missing, a directory, no permission...
logger.error("can't read %s: %s", path, error.strerror or error)
return 1
except ParseError as error:
logger.error("%s: %s", path, error)
return 1
records = [r for path in paths for r in load_records(conn, path) if keep(r)]
if args.json:
try:
save_summary(summary(conn), args.json)
except OSError as error:
logger.error("can't write %s: %s", args.json, error.strerror or error)
return 1
logger.info("wrote %s", args.json)
title = f"logan: {', '.join(paths)}"
print(Report(title, [
CountSection("Levels", key=lambda r: r.level),
CountSection("Busiest sources", key=lambda r: r.source, top=args.top),
ListSection("First errors", keep=level_at_least(Level.ERROR), limit=3),
]).render(records))
print(f"\n{len(records)} record(s) counted, {len(skipped)} bad line(s) skipped")
return 0
if __name__ == "__main__":
sys.exit(main())
# logan
Summarise server logs from the command line: how many records at each level,
which sources are busiest, and the first errors of the day.
```
$ python cli.py messy.log --level warning
logan: messy.log
================
Levels
ERROR 3
WARNING 1
...
```
## Log format
One record per line: date, time, level, source and a message of `key=value`
fields. Quoted values may contain spaces.
```
2026-09-01 09:15:30 ERROR payments: order=1043 msg="card declined"
```
Levels are DEBUG, INFO, WARNING, ERROR and CRITICAL, in any case. WARN, ERR
and FATAL are accepted as aliases. Everywhere in logan, an "error" means ERROR
or CRITICAL.
The `key=value` fields are extracted on a best-effort basis: text that isn't a
field is ignored, and a key that appears twice keeps its last value.
## Usage
```
python cli.py LOGFILE [LOGFILE ...] [--level LEVEL] [--source NAME] [--since HH:MM]
[--top N] [--json PATH] [--strict] [-v]
```
Needs Python 3.12 or later, and nothing outside the standard library.
| Option | Effect |
| --- | --- |
| `--level LEVEL` | count only records at LEVEL or worse |
| `--source NAME` | count only records from NAME; repeat for several |
| `--since HH:MM` | count only records from this time of day onwards |
| `--top N` | list the N busiest sources (default 5) |
| `--json PATH` | also write a JSON summary to PATH. It covers every record read, whatever the filters |
| `--strict` | stop at the first bad line instead of skipping it |
| `-v` | log progress to stderr |
The report goes to stdout. Warnings about skipped lines, and progress with
`-v`, go to stderr, so `python cli.py server.log > report.txt` keeps the report
clean.
## Exit codes
- `0`: the report was printed. Bad lines may have been skipped, and each one
was logged as a warning.
- `1`: a file couldn't be read, the JSON file couldn't be written, or
`--strict` met a bad line.
- `2`: the command line itself was wrong, e.g. an unknown level.
## Limits
- Lines are parsed as a stream, but each run keeps every record in an
in-memory database and holds the report's records in a list. logan is
meant for logs that fit comfortably in memory.
- Timestamps are assumed to be in one consistent format, without UTC offsets.
Mixing offsets would break time order.
## Design
- `analyzer.py` parses lines into `LogRecord`s and counts them. Bad lines raise
`ParseError`, or go to an `on_error` callback.
- `levels.py` has the `Level` enum. Levels are converted from text once, when
a line is parsed.
- `filters.py` builds predicates (`level_at_least`, `source_is`, `all_of`).
- `store.py` keeps records in SQLite. Importing a file twice stores it once.
- `report.py` renders reports made of sections.
- `cli.py` is the only module that knows about the command line, printing and
logging configuration.
## Development
Run the tests with `python -m unittest`. Type-check the typed modules with
`python -m mypy analyzer.py levels.py filters.py report.py`.
"""Log analyzer: 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 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")
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)
"""Keep parsed records in SQLite, and ask questions in SQL."""
import json
import logging
import sqlite3
from collections.abc import Iterable, Iterator
from contextlib import contextmanager
from datetime import datetime
from typing import Any
from analyzer import LogRecord, parse_fields
from levels import Level
logger = logging.getLogger(__name__)
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: str = ":memory:") -> sqlite3.Connection:
"""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: str = ":memory:") -> Iterator[sqlite3.Connection]:
"""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: sqlite3.Connection, log: str, records: Iterable[LogRecord]) -> int:
"""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)
logger.info("%s: stored %d new record(s)", log, cursor.rowcount)
return cursor.rowcount
def record_from_row(row: sqlite3.Row) -> LogRecord:
"""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: sqlite3.Connection, log: str) -> list[LogRecord]:
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: str) -> str:
"""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: sqlite3.Connection, text: str) -> list[LogRecord]:
"""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: Level | str) -> tuple[str, ...]:
"""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: sqlite3.Connection,
column: str,
level: Level | str | None = None,
*,
at_least: Level | str | None = None,
) -> dict[str, int]:
"""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: tuple[str, ...] = ()
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: sqlite3.Connection) -> dict[str, Any]:
"""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: dict[str, Any], path: str) -> None:
with open(path, "w") as file:
json.dump(data, file, indent=2)
def load_summary(path: str) -> Any:
with open(path) as file:
return json.load(file)
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 os
import unittest
from datetime import datetime
from analyzer import (ParseError, count, parse_fields, parse_lines, parse_record, read_lines,
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})
class TestReadLines(unittest.TestCase): # Day 7 edge cases
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)
def test_a_marked_first_line_parses(self): # Day 7 bug hunt
lines = self.read("\ufeff2026-09-01 09:14:02 INFO auth: user=ada\n".encode())
self.assertEqual(parse_record(lines[0]).source, "auth")
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 os
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.addCleanup(os.remove, "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})
import io
import json
import logging
import os
import unittest
from contextlib import redirect_stderr, redirect_stdout
from datetime import time
import cli
from cli import main
def run(*argv):
"""Run the command line in-process: (exit code, stdout, stderr)."""
root = logging.getLogger()
saved = root.handlers[:], root.level # main() configures logging: put it back after
out, err = io.StringIO(), io.StringIO()
try:
with redirect_stdout(out), redirect_stderr(err):
code = main(list(argv))
except SystemExit as exit: # argparse exits on --help and usage errors
code = exit.code
finally:
root.handlers[:], _ = saved
root.setLevel(saved[1])
return code, out.getvalue(), err.getvalue()
class TestCommandLine(unittest.TestCase):
def test_report_goes_to_stdout_and_warnings_to_stderr(self):
code, out, err = run("messy.log")
self.assertEqual(code, 0)
self.assertIn("6 record(s) counted, 2 bad line(s) skipped", out)
self.assertIn("skipped line 5", err)
self.assertNotIn("skipped line", out)
def test_level_and_source_filters(self):
code, out, _ = run("messy.log", "--level", "warn", "--source", "payments")
self.assertIn("2 record(s) counted", out)
def test_json_summary(self):
self.addCleanup(os.remove, "summary.json")
run("messy.log", "--json", "summary.json")
with open("summary.json") as file:
self.assertEqual(json.load(file)["records"], 6)
def test_verbose_logs_progress(self):
self.assertNotIn("stored 6", run("messy.log")[2])
self.assertIn("stored 6 new record(s)", run("messy.log", "-v")[2])
def test_strict_stops_at_the_first_bad_line(self):
code, _, err = run("messy.log", "--strict")
self.assertEqual(code, 1)
self.assertIn("line 5: bad timestamp", err)
def test_a_missing_file_is_an_error(self):
code, _, err = run("no-such.log")
self.assertEqual(code, 1)
self.assertIn("can't read no-such.log: No such file", err)
def test_a_directory_is_an_error_not_a_crash(self):
os.makedirs("a-directory", exist_ok=True)
self.addCleanup(os.rmdir, "a-directory")
code, _, err = run("a-directory")
self.assertEqual(code, 1)
self.assertIn("can't read a-directory", err)
def test_the_same_file_twice_counts_once(self):
_, out, _ = run("messy.log", "messy.log")
self.assertIn("6 record(s) counted, 2 bad line(s) skipped", out)
def test_json_summarises_every_record_whatever_the_filters(self):
self.addCleanup(os.remove, "summary.json")
_, out, _ = run("messy.log", "--level", "error", "--json", "summary.json")
self.assertIn("3 record(s) counted", out)
with open("summary.json") as file:
self.assertEqual(json.load(file)["records"], 6)
def test_an_unknown_level_is_a_usage_error(self):
code, _, err = run("messy.log", "--level", "loud")
self.assertEqual(code, 2)
self.assertIn("unknown level 'loud'", err)
class TestSince(unittest.TestCase): # the Day 7 checkpoint
def test_time_of_day(self):
self.assertEqual(cli.time_of_day("09:30"), time(9, 30))
self.assertEqual(cli.time_of_day("9:05"), time(9, 5))
def test_since_counts_only_later_records(self):
code, out, _ = run("messy.log", "--since", "09:16")
self.assertEqual(code, 0)
self.assertIn("3 record(s) counted", out)
def test_since_combines_with_level(self):
_, out, _ = run("messy.log", "--since", "09:16", "--level", "error")
self.assertIn("2 record(s) counted", out)
def test_a_bad_time_is_a_usage_error(self):
for text in ["24:00", "noon", "09:30:15"]:
with self.subTest(text=text):
code, _, err = run("messy.log", "--since", text)
self.assertEqual(code, 2)
self.assertIn(f"expected a time like 09:30, not {text!r}", err)
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"
Look back at where it started: the 27-line report.py from Day 1. Everything
it counted, this still counts. The characterization test from that first
morning has gone, replaced by better tests, one piece at a time. On top of
that, the analyzer now:
parses each line once into a typed, frozen record, and never drops a line
without saying so,
parses logs as a stream, line by line, whatever their encoding quirks,
keeps records in SQLite, safely, and can import the same log twice,
exports JSON that survives the round trip,
takes its filters, callbacks and report sections as parts to combine,
knows its levels as an ordered type, not as spellings,
logs diagnostics to stderr, and has a command line with honest exit codes,
passes mypy --strict, and has a regression test for the bugs it met along
the way.
That list has a limit worth stating. The parser streams, but each run then keeps
every record in an in-memory database and the report’s records in a list, so
the finished tool is for logs that fit comfortably in memory. Its README says
so, under “Limits”. Handling logs bigger than memory would be a different
design, not a missing feature.
Run it on your machine
Download the finished analyzer,
unzip it, and open a terminal in its folder. You need Python 3.12 or later, and
nothing outside the standard library. Then:
python cli.py messy.log --level warning
python -m unittest
python -m unittest finds every test_*.py in the folder by itself, so you
don’t need run_tests.py or main.py on your machine.
Going further: installing it as a command
Optional, and on your own machine
Nothing in the course depends on this section.
Typing python cli.py works, but a real tool is a command you can run from
anywhere. Add a pyproject.toml beside the modules:
[build-system]requires = ["setuptools>=69"]build-backend = "setuptools.build_meta"[project]name = "logan"version = "1.0.0"description = "Summarise server logs from the command line."readme = "README.md"requires-python = ">=3.12"[project.scripts]logan = "cli:main"[tool.setuptools]py-modules = ["analyzer", "cli", "filters", "levels", "report", "store"]
Then, inside a virtual environment:
python -m pip install -e .
logan messy.log --level error
[project.scripts] creates the logan command, which calls cli.main() and
exits with whatever it returns. That’s one more reason main returns an exit
code. -e installs it in editable mode, so changes to the code take effect
without reinstalling. A larger project would move its modules into a package
directory, logan/. With six modules, a flat layout is fine.
What you can now do
Add a feature test-first, and let the existing tests prove nothing else
changed.
Write a README that answers a newcomer’s questions, with real output in it.
Ship a tool: tested, typed, logged, documented and runnable as a command.
The week, looking back
Seven days ago, the analyzer was a script that worked and that nobody dared to
touch. Each day added one habit that makes code survive contact with real
input and real colleagues:
Change code under a safety net, in small steps.
Parse once, at the edge, and reject bad input loudly and precisely.
Store and exchange data without losing its meaning on the way.
Pass behaviour in as functions, and clean up on every path.
Compose small parts, and let types say what the code assumes.
Wait concurrently, with limits, and handle failure per source.
Finish the job: edge cases, logging, a command line and a README.
And every day, the same routine: reproduce, inspect, narrow, pin, fix. You’ll
use it long after you’ve forgotten which module argparse lives in.
Two more pages
logan was built with a lot of support: a plan for every day, and tests
written for you. Two short pages finish the course:
On Your Own — a fresh brief in a new
domain, where you design the functions and write the tests.
Come Back Later — review
challenges for a week from now and a month from now.
Where next
There’s no single next step; it depends on what you want to build.
If you want to…
Try
Point this at your own data
Rebuild logan’s shape for a log or export you actually have
Talk to real services
httpx with Day 6’s timeouts and per-source failures, for real
Share a tool
Package it (Going further, above) and install it with pipx
Analyse data
pandas and Jupyter — you already know how to test what you compute
Build a web backend
FastAPI, which is built on the type hints and async you used this week
The advanced course, Designing Reliable Libraries, is one of those
destinations: it’s for people who write code that other developers call, and
decide what its API looks like. It isn’t the step everyone should take next —
if you want to build tools or analyse data, the other rows are the better move.
This course is being piloted. If you worked through it, what took longer than
expected, and where you got stuck, is the most useful thing you could share.
Your progress is saved in this browser.
Sign in
to keep it on every device.