Checkpoint: A Configurable Analyzer

Today you took the decisions out of the analyzer and handed them to its caller: what to count by, which records to count, what to do with a bad line. The analyzer does the parsing and counting. The caller supplies the judgement, as functions.

The checkpoint puts all of that together, and adds the one piece of the day the analyzer hasn’t used yet: a context manager of your own.

Day 4 checkpoint

from contextlib import contextmanager

from analyzer import count, parse_lines


def analyze(lines, *, key, keep=None, on_error=None):
    """Parse lines and count the records by key(record), keeping only those keep() accepts.

    Bad lines go to on_error; with no on_error, the first one is raised.
    """
    return count(parse_lines(lines, on_error), key, keep)


def field_equals(name, value):
    """A filter: keep records whose message has the field name=value."""
    def keep(record):
        return record.fields.get(name) == value

    return keep


@contextmanager
def error_log(path):
    """Open path for writing and give the with block an on_error callback that
    writes each error to it, one per line. The file is closed however the block ends."""
    with open(path, "w") as file:
        def write(error):
            file.write(f"{error}\n")

        yield write

The Day 4 reference solution

The analyzer at the end of Day 4, and where Day 5 starts. It has a new module, filters.py, which holds today’s filter factories. parse_lines takes on_error, analyzer.py gained count, and store.py gained open_store. Twenty-eight tests, including one for each of today’s three bug hunts, and Day 3’s tests that a failed save stores nothing.

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

Going further: decorators

Today’s course plan said: a decorator only if something genuinely repeats. The analyzer doesn’t need one yet, and that’s a fine outcome. But here’s what it would look like if it did.

Suppose you want to know how long each stage takes. You’d write start = time.perf_counter() before a call and print(...) after it, around the parse, around the save, and around every count. When the same wrapping code surrounds many different functions, a decorator can hold it in one place:

import functools
import time


def timed(function):
    """Print how long each call of function takes."""
    @functools.wraps(function)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            return function(*args, **kwargs)
        finally:
            elapsed = time.perf_counter() - start
            print(f"{function.__name__} took {elapsed * 1000:.1f} ms")

    return wrapper


@timed
def total(n):
    """Add up the numbers below n."""
    return sum(range(n))


@timed
def squares(n):
    """Yield the squares below n, one at a time."""
    for i in range(n):
        yield i * i


print(total(2_000_000))
print(total.__name__, "-", total.__doc__)

numbers = squares(2_000_000)          # "took 0.0 ms": nothing has run yet
print(sum(numbers))

A decorator is a function factory like the ones from this morning: it takes a function and returns a new one. @timed above def total is shorthand for total = timed(total).

Three details matter:

• *args, **kwargs let the wrapper accept whatever the original accepts, and pass it all on unchanged.
• functools.wraps copies the original’s name and docstring onto the wrapper. Remove that line and total.__name__ becomes wrapper, which makes every traceback, log line and help() harder to read.
• try/finally means the timing is printed even when the function raises. It’s the same rule as context managers.

Now look at squares: “took 0.0 ms” for two million squares. Calling a generator function only creates the generator, and the work happens later, when sum reads it. It’s the same surprise as this morning’s callback, and a reminder that wrapping code assumes something about when the work happens. For code like this, a context manager around the whole stage often measures what you actually meant.

A decorator earns its place when the same concern, like timing, retrying or logging, wraps many functions that have nothing else in common. For one or two call sites, the plain code is clearer.

What you can now do

• Combine callbacks, filter factories and existing functions into small, configurable APIs.
• Write a context manager whose cleanup is proven by a test on the failure path.
• Recognise when a decorator would remove real repetition, and write one with functools.wraps, or decide not to.

Tomorrow: the analyzer gets a report made of parts, type hints on its public functions, and a bug that no test caught because every test agreed with it.