Functions as Arguments

Warm-up: from Day 3

From memory first.

1. Predict both lines — the second is an injection attempt.

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE t (level TEXT)")
conn.executemany("INSERT INTO t VALUES (?)", [("ERROR",), ("INFO",), ("ERROR",)])

print(conn.execute("SELECT level, COUNT(*) FROM t GROUP BY level ORDER BY level").fetchall())
print(conn.execute("SELECT COUNT(*) FROM t WHERE level = ?", ("x' OR '1'='1",)).fetchone())

[('ERROR', 2), ('INFO', 1)]
(0,)

2. Fix. Status counts saved as JSON and loaded back should equal what you saved: this should print True 5. It crashes.

import json

counts = {200: 5, 404: 1}
restored = json.loads(json.dumps(counts))
print(restored == counts, restored[200])

import json

counts = {200: 5, 404: 1}
data = [{"status": status, "count": n} for status, n in counts.items()]
restored = {item["status"]: item["count"] for item in json.loads(json.dumps(data))}
print(restored == counts, restored[200])

True 5

3. Explain. No code: count_by(conn, column, level) defends against a hostile column and a hostile level in two different ways. What are they, and why can’t it use the same one for both?

Where Day 3 ended

Here’s where Day 3 ended: records parsed once, stored in SQLite, and summarised as JSON.

People have started asking the analyzer questions nobody planned for. Errors per hour. Warnings per source. Payment problems after 10:00. Each new question so far has meant a new function, and they’re starting to look alike.

Three loops, one idea

Here are three counting functions from earlier in the week, side by side:

def count_levels(records):
    counts = Counter()
    for r in records:
        counts[r.level] += 1
    return counts


def errors_by_source(records):
    counts = Counter()
    for r in records:
        if r.level == "ERROR":
            counts[r.source] += 1
    return counts


def errors_by_hour(records):
    counts = Counter()
    for r in records:
        if r.level == "ERROR":
            counts[r.timestamp.hour] += 1
    return counts

Only two things differ: what to count by (r.level, r.source, r.timestamp.hour) and which records to include (all of them, or just the errors). Everything else is the same loop, copied.

When the part that varies is behaviour rather than a value, make it a function, and pass it in:

from collections import Counter
from dataclasses import dataclass


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


RECORDS = [
    LogRecord(9, "INFO", "auth"),
    LogRecord(9, "ERROR", "payments"),
    LogRecord(9, "ERROR", "search"),
    LogRecord(10, "ERROR", "payments"),
    LogRecord(10, "WARNING", "payments"),
]


def count(records, key, keep=None):
    """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 is_error(record):
    return record.level == "ERROR"


print(count(RECORDS, key=lambda r: r.level))
print(count(RECORDS, key=lambda r: r.source, keep=is_error))
print(count(RECORDS, key=lambda r: r.hour, keep=is_error))

Three functions became one, and the fourth question, and the fortieth, cost one line each instead of a new function.

This is the same idea as sorted(records, key=...), which you met in the beginner course. sorted knows how to sort. You tell it what to sort by. count knows how to count, and you tell it what to count by and what to count.

lambda or def?

key=lambda r: r.source and keep=is_error are both functions. The only difference is that one has a name.

• Use lambda for a short expression that means nothing outside this one call, like lambda r: r.source.
• Use def as soon as it has a meaning of its own, is used twice, needs a docstring or a test, or wouldn’t fit comfortably on one line. is_error reads better at the call site than lambda r: r.level == "ERROR", and a typo in it gets fixed in one place.

Note keep=is_error, not keep=is_error(). You’re handing over the function itself, for count to call later, once per record. With brackets you’d call it now, with no record, and get a TypeError.

Letting the caller decide: on_error

On Day 2, parse_lines took a list and appended each bad line’s ParseError to it. The bug hunt showed how easy that was to misuse: pass [], and the errors are collected and thrown away. The lesson promised a better design, and here it is. Instead of a list, parse_lines takes a function, on_error, and calls it with each error:

def parse_lines(lines, on_error=None):
    """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

A function you pass in for someone else to call back later is a callback. This one lets a single parse_lines support every strategy from Day 2, without knowing which one it’s serving:

errors.append is a bound method: a function that already knows which list it appends to. Passing it is how you say “put the errors in this list”.

The most important line is the default. Forgetting on_error doesn’t drop errors. It stops at the first one. Day 2’s silent drop needed nothing more than a missing name, and now it needs a deliberate on_error=lambda error: None, which no one writes by accident.

Bug hunt: the report that saw no errors

The report below uses the new on_error. It says there were no bad lines, but there’s one in the input, and the levels add up to only four of the five lines:

from collections import Counter


def parse_lines(lines, on_error=None):
    for number, line in enumerate(lines, start=1):
        parts = line.split()
        if len(parts) < 4:
            if on_error is None:
                raise ValueError(f"line {number}: too few parts")
            on_error(f"line {number}: too few parts")
        else:
            yield parts[2]          # just the level, to keep the example short


LINES = [
    "2026-09-01 09:14:02 INFO    auth: user=ada",
    "2026-09-01 09:15:30 ERROR   payments: order=1043",
    "garbage",
    "2026-09-01 09:17:45 ERROR   search: query=dune",
    "2026-09-01 09:18:02 INFO    search: query=dune",
]

errors = []
levels = parse_lines(LINES, on_error=errors.append)
print(f"Skipped {len(errors)} bad line(s)")
print("Levels:", dict(Counter(levels)))

Reproduce. Five lines, one of them garbage. The report says it skipped 0.

Inspect. Print the state that’s wrong, at the end rather than where it’s used. Add print(errors) as the very last line. It prints ['line 3: too few parts']. So the error was collected. It just wasn’t there yet when the report printed its count.

Before you open this, explain in one sentence why errors was still empty when the count was printed. Then fix the report and compare.

def test_errors_are_known_once_the_records_are_read(self):
    errors = []
    records = parse_lines(["garbage", GOOD], on_error=errors.append)
    self.assertEqual(errors, [])            # a generator: nothing has run yet
    list(records)
    self.assertEqual(len(errors), 1)

Your turn

import sys
import unittest
from collections import Counter
from dataclasses import dataclass


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


def count(records, key, keep=None):
    raise NotImplementedError("your code here")


def group(records, key):
    raise NotImplementedError("your code here")


RECORDS = [
    LogRecord(9, "INFO", "auth"),
    LogRecord(9, "ERROR", "payments"),
    LogRecord(10, "ERROR", "search"),
    LogRecord(10, "ERROR", "payments"),
]


class TestFunctionsAsArguments(unittest.TestCase):
    def test_count_everything(self):
        self.assertEqual(count(RECORDS, key=lambda r: r.level), {"INFO": 1, "ERROR": 3})

    def test_count_some(self):
        self.assertEqual(count(RECORDS, key=lambda r: r.hour, keep=lambda r: r.source == "payments"),
                         {9: 1, 10: 1})

    def test_group_keeps_the_original_order(self):
        groups = group(RECORDS, key=lambda r: r.hour)
        self.assertEqual(list(groups), [9, 10])
        self.assertEqual(groups[10], [RECORDS[2], RECORDS[3]])

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

        hour, _ = count(RECORDS, key=lambda r: r.hour, keep=is_error).most_common(1)[0]
        sources = [r.source for r in group(RECORDS, key=lambda r: r.hour)[hour] if is_error(r)]
        self.assertEqual((hour, sources), (10, ["search", "payments"]))


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

def count(records, key, keep=None):
    """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 group(records, key):
    """{key(record): [records with that key]}, each list in the original order."""
    groups = {}
    for record in records:
        groups.setdefault(key(record), []).append(record)
    return groups

What you can now do

• Spot the part of near-duplicate code that is behaviour, and pass it in as a function.
• Choose between lambda and def, and pass a function without calling it.
• Design a callback with a safe default, like on_error failing fast unless told otherwise.
• Remember that a callback passed to a generator runs only when the generator is read.

Next: functions that make functions, so keep= arguments read like English.