Project: a log analyzer from start to finish

Why this matters

Every running service writes logs, and when something breaks, the first question is "what does the log say?" Scrolling through ten thousand lines does not scale; a small tool that parses, filters and summarizes does. This project builds such a tool the way experienced engineers build any program: define the input precisely, split the work into small testable functions, handle bad data deliberately, and wrap it in a proper command-line interface. The same structure works for nearly every data-processing script you will write.

Step 0: specify before coding

Start with a sample of the real input and decide what counts as valid:

2026-10-31T12:00:01 INFO web: server started on :8080
2026-10-31T12:01:12 ERROR auth: invalid password for user bob
2026-10-31T12:01:15 WARNING web: slow response: 2.3s
this line is garbage

The format: TIMESTAMP LEVEL COMPONENT: MESSAGE. Questions to answer before writing code, because real logs will test every one:

• Can the message contain colons? (Yes: slow response: 2.3s. So split on the first ": " only.)
• Can it contain spaces? (Yes. So split on spaces at most twice.)
• What about lines that do not match? (Real logs contain stack traces, truncated lines, output from other programs. We will skip and count them rather than crash, and report the count so problems are visible.)
• Which levels exist, and in what order of severity? (DEBUG < INFO < WARNING < ERROR < CRITICAL, the same names Python's logging module uses.)

Then decide on the output. A good report is compact and stable:

lines: 10 (skipped 2)
DEBUG: 1
INFO: 3
WARNING: 1
ERROR: 2
CRITICAL: 1
errors by component:
  auth: 2
  disk: 1

The architecture

 file ──> read_entries ──> parse_line (per line) ──> [Entry, ...]
                                                         │
                       args (--level, --top) ──> summarize ──> Summary
                                                         │
                                                format_report ──> str ──> print

Only main touches the outside world: arguments, files, stdout, stderr and exit codes. Everything else takes values and returns values. This "functional core, imperative shell" design means each stage can be tested in isolation with a string or a list, and the core could be reused to produce JSON or feed a dashboard without changes.

Step 1: parsing one line

Represent a parsed line as a small immutable record:

from dataclasses import dataclass

LEVELS = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]

@dataclass(frozen=True)
class Entry:
    timestamp: str
    level: str
    component: str
    message: str

A dataclass gives readable attribute names (e.level rather than e[1]), a useful repr for debugging, and equality, so tests can compare whole entries. frozen=True makes entries hashable and protects them from accidental modification.

Parsing uses two string tools with a limit:

>>> "2026-10-31T12:01:15 WARNING web: slow response: 2.3s".split(" ", 2)
['2026-10-31T12:01:15', 'WARNING', 'web: slow response: 2.3s']
>>> "web: slow response: 2.3s".partition(": ")
('web', ': ', 'slow response: 2.3s')

split(" ", 2) performs at most two splits, so the remainder stays intact. partition splits on the first occurrence only and always returns three parts; the middle one is empty when the separator is missing, which makes the "malformed" check simple. For the timestamp, datetime.fromisoformat validates for us: it raises ValueError for 2026-13-45T... or yesterday.

Returning None for malformed lines (rather than raising) is a design decision: in this tool a bad line is expected, not exceptional. A function used elsewhere for strict validation might raise a custom ParseError instead. Whichever you choose, make it explicit in the type hint: -> Entry | None.

Step 2: summarizing

The summary needs counts per level, restricted to a minimum level, and the components with the most errors. Severity order comes free from the list: LEVELS.index("WARNING") is 2, so LEVELS[2:] are the levels to report.

from collections import Counter

errors = Counter()
for e in entries:
    if e.level in ("ERROR", "CRITICAL"):
        errors[e.component] += 1
ranked = sorted(errors.items(), key=lambda kv: (-kv[1], kv[0]))

Two details make the output trustworthy. First, every reported level appears even with a count of zero, so a reader can tell "no errors" from "errors not reported". Second, ties are broken by name: Counter.most_common would order ties by first appearance, which changes whenever the log changes slightly and makes test output flaky.

summarize accepts any iterable of entries and loops over it exactly once. That means it works on a list, but also on a generator that parses a 5 GB log lazily, line by line, without holding it all in memory:

entries = (e for line in f if (e := parse_line(line)) is not None)
summary = summarize(entries)

Step 3: the shell around the core

main wires it together:

def main() -> int:
    args = build_parser().parse_args()
    try:
        with open(args.logfile, encoding="utf-8") as f:
            entries, lines, skipped = read_entries(f)
    except OSError:
        print(f"error: cannot read {args.logfile}", file=sys.stderr)
        return 1
    summary = summarize(entries, args.level, args.top)
    print(format_report(summary, lines, skipped))
    return 0

if __name__ == "__main__":
    sys.exit(main())

• argparse validates --level with choices=LEVELS and --top with type=int; bad usage exits with status 2 automatically.
• OSError is the parent of FileNotFoundError, IsADirectoryError and PermissionError, so one handler covers every way opening can fail. The message goes to stderr and the exit status is 1, so scripts calling the tool can detect failure (analyzer app.log || alert).
• format_report returns a string instead of printing. Tests can compare it directly, and main decides where it goes.

Real-world logs sometimes contain invalid UTF-8 bytes. Opening with errors="replace" turns them into � replacement characters instead of crashing with UnicodeDecodeError, a sensible choice for a diagnostic tool.

Testing the whole thing

Because the core is pure, a unittest module for it is short:

import unittest
from analyzer import parse_line, summarize, Entry

class ParseTests(unittest.TestCase):
    def test_message_with_colons(self):
        e = parse_line("2026-10-31T12:00:02 INFO web: GET /a: 200")
        self.assertEqual(e, Entry("2026-10-31T12:00:02", "INFO", "web", "GET /a: 200"))

    def test_garbage_is_none(self):
        self.assertIsNone(parse_line("this line is garbage"))

class SummaryTests(unittest.TestCase):
    def test_zero_counts_present(self):
        s = summarize([], min_level="ERROR")
        self.assertEqual(s.levels, {"ERROR": 0, "CRITICAL": 0})

if __name__ == "__main__":
    unittest.main()

The CLI layer is then tested end to end by running the program on small sample files and comparing output and exit status, exactly what learnprog's checks do for the final exercise.

Common mistakes

• ValueError: not enough values to unpack (expected 3, got 1) from ts, level, rest = line.split(" ", 2) on a garbage line: check the length (or use partition) before unpacking.
• Splitting the message: rest.split(": ") without a limit breaks messages containing ": " and makes the unpacking fail with too many values to unpack.
• Forgetting the newline: the last field keeps a trailing "\n" unless you strip the line.
• Counting blank lines as malformed: decide on a rule and keep it consistent; here blank lines are ignored entirely.
• Order depending on input order: sort explicitly with a tie-breaker.
• Catching Exception around everything in main: it hides programming errors (a typo becomes "cannot read file"). Catch the specific errors you expect.

Recap

• Specify the input format and the edge cases (colons in messages, garbage lines, levels) before coding.
• Structure the program as pure functions (parse_line, summarize, format_report) with all I/O in main.
• Use dataclasses for records, split with a limit and partition for parsing, Counter plus an explicit sort for ranking.
• Skip and count bad input rather than crashing; report errors on stderr with a nonzero exit status; let argparse handle usage errors.
• Pure cores are easy to unit-test; the CLI is tested end to end with sample files.

Going further: extend the analyzer with a --since TIMESTAMP option (compare datetime.fromisoformat values), and a --json flag that prints the summary with json.dumps(dataclasses.asdict(summary), indent=2), reusing the same core functions.