Reports Made of Parts

Warm-up: from Day 4

From memory first.

1. Predict both lines. The second is Day 4’s bug hunt in miniature.

def make_multiplier(n):
    def times(x):
        return x * n
    return times

double = make_multiplier(2)
triple = make_multiplier(3)
print(double(5), triple(5))

adders = [lambda x: x + i for i in range(3)]
print([add(10) for add in adders])

10 15
[12, 12, 12]

2. Fix. The log should show the context closing even when the block fails: ['open report', 'close report'].

from contextlib import contextmanager

log = []

@contextmanager
def tracked(name):
    log.append(f"open {name}")
    yield name
    log.append(f"close {name}")

try:
    with tracked("report"):
        raise ValueError("boom")
except ValueError:
    pass

print(log)

from contextlib import contextmanager

log = []

@contextmanager
def tracked(name):
    log.append(f"open {name}")
    try:
        yield name
    finally:
        log.append(f"close {name}")

try:
    with tracked("report"):
        raise ValueError("boom")
except ValueError:
    pass

print(log)

['open report', 'close report']

3. Write count(items, key), returning a Counter of key(item). This should print {3: 2, 5: 1}:

from collections import Counter

# def count(items, key): ...

print(dict(count(["cat", "dog", "horse"], key=len)))

from collections import Counter

def count(items, key):
    return Counter(key(item) for item in items)

print(dict(count(["cat", "dog", "horse"], key=len)))

{3: 2, 5: 1}

Where Day 4 ended

Here’s where Day 4 ended: an analyzer configured by functions you pass in.

The questions have become reports. The on-call engineer wants errors by source and by hour. The payments team wants their own warnings and a list of the latest failures. A manager wants one line: how many records, how many errors. Each is a few counts and lists under headings, in a different combination.

The inheritance trap

A first instinct is often a class hierarchy: a base Report that prints a title, and subclasses that add parts.

class Report:                        # title and underline
class ErrorReport(Report):           # + errors by source
class HourlyErrorReport(ErrorReport):    # + errors by hour
class PaymentsReport(Report):        # + payments warnings
class PaymentsHourlyReport(...):     # errors by hour, but only payments... which parent?

It works for the first two reports. By the fifth, each new combination needs a new class, parents get changed to suit one child, and “which class prints the hourly part?” takes a search. Inheritance is good at “a kind of” relationships, and a report isn’t a kind of another report. It’s made of parts.

Composition: a report has sections

So model exactly that. A section is a small object with a title and a render(records) method that returns its lines. A report is a title plus a list of sections, and it renders by asking each section in turn:

from collections import Counter
from dataclasses import dataclass


@dataclass(frozen=True)
class LogRecord:
    hour: int
    level: str
    source: str
    message: str


RECORDS = [
    LogRecord(9, "INFO", "auth", "user=ada action=login"),
    LogRecord(9, "ERROR", "payments", "order=1043 card declined"),
    LogRecord(9, "ERROR", "search", "index offline"),
    LogRecord(10, "ERROR", "payments", "order=1050 timeout"),
]


@dataclass(frozen=True)
class CountSection:
    """Records counted by key, busiest first."""
    title: str
    key: object              # a function: record -> what to count by
    keep: object = None      # a function: record -> bool, or None for every record

    def render(self, records):
        counts = Counter(self.key(r) for r in records if self.keep is None or self.keep(r))
        return [f"  {name!s:<12}{n:>4}" for name, n in counts.most_common()] or ["  (none)"]


@dataclass(frozen=True)
class ListSection:
    """The first few records a filter keeps."""
    title: str
    keep: object
    limit: int = 5

    def render(self, records):
        chosen = [r for r in records if self.keep(r)][: self.limit]
        return [f"  {r.hour:02}:00 {r.source}: {r.message}" for r in chosen] or ["  (none)"]


@dataclass(frozen=True)
class Report:
    title: str
    sections: list

    def render(self, records):
        lines = [self.title, "=" * len(self.title)]
        for section in self.sections:
            lines += ["", section.title, *section.render(records)]
        return "\n".join(lines)


def is_error(record):
    return record.level == "ERROR"


on_call = Report("On-call summary", [
    CountSection("Errors by source", key=lambda r: r.source, keep=is_error),
    CountSection("Errors by hour", key=lambda r: r.hour, keep=is_error),
    ListSection("First errors", keep=is_error, limit=2),
])
print(on_call.render(RECORDS))

Every report is now just data: a title and a list. The payments team’s report is a different list of the same parts, with no new class. And each part is a frozen dataclass whose fields are exactly its settings, so a section is easy to read, compare and test on its own.

Two small details:

• [... ] or [" (none)"] uses the fact that an empty list is falsy, so a section with nothing to show still says so, instead of leaving a heading with no lines under it.
• [..., *section.render(records)] unpacks the section’s lines into the list being built.

Anything that renders is a section

Report.render never checks what kind of section it has. It only calls section.title and section.render(records). Any object with those two works, including a class written long after Report:

@dataclass(frozen=True)
class NoteSection:
    title: str
    text: str

    def render(self, records):
        return [f"  {self.text}"]

Adding it to a report changes nothing in Report. Python’s name for this is duck typing: if it has a title and a render, it’s a section. The next lesson, on type hints, shows how to write that expectation down, so a checker can hold you to it.

Bug hunt: the sections that went quiet

Here’s the analyzer’s first real report.py, used for the daily report. The first section is right. The next two say (none), and messy.log certainly has warnings and errors:

Reproduce. Every time. The first section always works, and every section after it is empty, whichever order you put them in. Try swapping two.

Inspect. What does each section receive? Add a line at the top of CountSection.render in report.py:

        print("render got", type(records).__name__)

It prints render got generator.

Your turn, before you open this. The input is already small, so start with the test: write one that renders a two-section report from a generator and fails now. Then fix Report, not the sections, and compare with ours.

def test_every_section_sees_every_record(self):
    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)

    def render(self, records):
        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)

Your turn

import sys
import unittest
from dataclasses import dataclass


@dataclass(frozen=True)
class LogRecord:
    level: str
    source: str


# Write ShareSection here.


RECORDS = [LogRecord("INFO", "auth"), LogRecord("ERROR", "db"),
           LogRecord("ERROR", "db"), LogRecord("WARNING", "auth")]


class TestShareSection(unittest.TestCase):
    def test_share(self):
        section = ShareSection("Errors", keep=lambda r: r.level == "ERROR")
        self.assertEqual(section.render(RECORDS), ["  2 of 4 (50.0%)"])

    def test_one_decimal_place(self):
        section = ShareSection("From auth, not INFO",
                               keep=lambda r: r.source == "auth" and r.level != "INFO")
        self.assertEqual(section.render(RECORDS[:3]), ["  0 of 3 (0.0%)"])

    def test_no_records(self):
        section = ShareSection("Errors", keep=lambda r: r.level == "ERROR")
        self.assertEqual(section.render([]), ["  0 of 0"])

    def test_it_has_a_title_like_every_section(self):
        self.assertEqual(ShareSection("Errors", keep=bool).title, "Errors")


suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestShareSection)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)

@dataclass(frozen=True)
class ShareSection:
    """What share of the records keep() accepts."""
    title: str
    keep: object

    def render(self, records):
        total = len(records)
        kept = sum(1 for r in records if self.keep(r))
        if total == 0:
            return ["  0 of 0"]
        return [f"  {kept} of {total} ({kept / total:.1%})"]

What you can now do

• Recognise when a class hierarchy is standing in for “made of”, and use composition instead.
• Build a report from small frozen-dataclass parts that are configured by data.
• Rely on duck typing: anything with the right methods can take part.
• Decide, in one place, when a stream must become a list.

Next: writing down what each function expects and returns, with type hints.