Functions That Build Functions

count(records, key=..., keep=...) can answer any question, but the keep functions are getting hard to read:

count(records, key=lambda r: r.source,
      keep=lambda r: LEVEL_ORDER.index(r.level) >= LEVEL_ORDER.index("WARNING"))

That’s “warnings and worse, by source”, but you have to decode it to find out. And the next person who needs “errors and worse” will copy the lambda and change one word, which is how Day 1’s script got the way it was.

What you want to write is this:

count(records, key=lambda r: r.source, keep=level_at_least("WARNING"))

level_at_least("WARNING") has to return a function, one that takes a record and answers yes or no.

A function factory

level_at_least Python

from dataclasses import dataclass

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


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


def level_at_least(minimum):
    """Keep records at minimum or more severe: level_at_least("WARNING")."""
    rank = LEVEL_ORDER.index(minimum)        # an unknown level fails now, not later

    def keep(record):
        return LEVEL_ORDER.index(record.level) >= rank

    return keep


warnings_or_worse = level_at_least("WARNING")
print(warnings_or_worse)
print(warnings_or_worse(LogRecord("ERROR", "db")))
print(warnings_or_worse(LogRecord("INFO", "db")))

errors_or_worse = level_at_least("ERROR")
print(errors_or_worse(LogRecord("WARNING", "db")))

level_at_least("WARN")

Each call to level_at_least runs its body once. It works out rank, defines a new keep function, and returns it. The returned keep still uses rank, even though level_at_least has finished and its local variables would normally be gone.

A function that keeps using variables from the function it was defined in is called a closure. It closes over those variables and keeps them alive for as long as it exists. warnings_or_worse and errors_or_worse are two separate keep functions, each with its own rank.

Two design details are worth copying:

Work that only depends on the argument happens once. LEVEL_ORDER.index(minimum) runs when the filter is built, not once per record.
Bad arguments fail when the filter is built. level_at_least("WARN") raises ValueError straight away, on the line with the typo. If the check were inside keep, it would fail somewhere deep in a report, later, far from the mistake.

Combining filters

Functions that take filters and return a filter let you build questions out of small parts:

all_of, and friends Python

from dataclasses import dataclass
from datetime import datetime

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


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


def level_at_least(minimum):
    rank = LEVEL_ORDER.index(minimum)

    def keep(record):
        return LEVEL_ORDER.index(record.level) >= rank

    return keep


def source_is(*sources):
    """Keep records from any of the given sources."""
    wanted = set(sources)

    def keep(record):
        return record.source in wanted

    return keep


def all_of(*predicates):
    """Keep records that every predicate keeps."""
    def keep(record):
        return all(predicate(record) for predicate in predicates)

    return keep


RECORDS = [
    LogRecord(datetime(2026, 9, 1, 9, 15), "ERROR", "payments"),
    LogRecord(datetime(2026, 9, 1, 9, 16), "INFO", "payments"),
    LogRecord(datetime(2026, 9, 1, 9, 17), "WARNING", "auth"),
    LogRecord(datetime(2026, 9, 1, 10, 2), "WARNING", "payments"),
]

payment_problems = all_of(source_is("payments"), level_at_least("WARNING"))
for record in filter(payment_problems, RECORDS):
    print(record)

payment_problems reads like its name, and each part can be tested on its own. The built-in filter(predicate, items) accepts any of these filters too, because to Python they’re just functions that take one argument.

Bug hunt: every filter the same

The command line will let people pick a minimum level by name, so someone builds all five filters up front, in a dictionary, from a loop:

One filter per level Python

from dataclasses import dataclass

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


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


AT_LEAST = {
    level: lambda record: LEVEL_ORDER.index(record.level) >= LEVEL_ORDER.index(level)
    for level in LEVEL_ORDER
}

info = LogRecord("INFO")
for name, keep in AT_LEAST.items():
    print(f"at least {name:8} keeps INFO? {keep(info)}")

Reproduce. “At least DEBUG” should keep an INFO record. So should “at least INFO”. Every filter says False.

Inspect. Every filter gives the same answer, so find out which answer. Try the “at least DEBUG” filter on a record at each level:

print([AT_LEAST["DEBUG"](LogRecord(level)) for level in LEVEL_ORDER])

It keeps only CRITICAL. The filter made for DEBUG behaves exactly like the one made for CRITICAL, the last one the loop made.

(Optional, if you’re curious: a closure keeps the variables it uses in its __closure__, and keep.__closure__[0].cell_contents shows the current value. It’s 'CRITICAL' for all five filters.)

Before you open this: which value of level does every lambda see when it runs, and why? Then fix it using something from earlier in this lesson, not a new trick, and compare.

Show the diagnosis and the fix

Narrow. A closure doesn’t copy a variable’s value when it’s created. It keeps a reference to the variable, and looks up its value each time it runs. All five lambdas were made in the same loop, so they share the loop’s one level variable. By the time any of them runs, the loop has finished, and level holds its last value, "CRITICAL".

Pin it. A test that builds filters in a loop and checks two of them disagree:

def test_filters_made_in_a_loop_are_all_different(self):
    by_level = {level: level_at_least(level) for level in LEVEL_ORDER}
    self.assertTrue(by_level["DEBUG"](record("INFO")))
    self.assertFalse(by_level["ERROR"](record("INFO")))

Fix. Give each function its own variable, by making each one in its own call of a factory. That’s exactly what level_at_least does:

AT_LEAST = {level: level_at_least(level) for level in LEVEL_ORDER}

Each call of level_at_least has its own minimum and its own rank, so each keep closes over different variables. Paste level_at_least from earlier in the lesson above the dictionary, change the dictionary to this, and run again.

The default-argument trick

You’ll also see lambda record, level=level: ... in older code. A default is evaluated once, when the function is defined, so each lambda gets the loop’s value at that moment. It works, but it adds a parameter any caller could override by accident. A factory function says what it means.

Your turn

Exercise

Three more filter builders

Write three factories that return filters:

any_of(*predicates) keeps a record if any predicate keeps it.
negate(predicate) keeps exactly the records predicate doesn’t.
message_contains(text) keeps records whose message contains text, ignoring case.

import sys
import unittest
from dataclasses import dataclass


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


def any_of(*predicates):
    raise NotImplementedError("your code here")


def negate(predicate):
    raise NotImplementedError("your code here")


def message_contains(text):
    raise NotImplementedError("your code here")


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


def from_db(record):
    return record.source == "db"


TIMEOUT = LogRecord("ERROR", "db", "Query TIMEOUT after 30s")
LOGIN = LogRecord("INFO", "auth", "user=ada action=login")
LAG = LogRecord("WARNING", "db", "replica lag 4s")


class TestFilterBuilders(unittest.TestCase):
    def test_any_of(self):
        keep = any_of(is_error, from_db)
        self.assertEqual([keep(r) for r in (TIMEOUT, LOGIN, LAG)], [True, False, True])

    def test_any_of_nothing_keeps_nothing(self):
        self.assertFalse(any_of()(TIMEOUT))

    def test_negate(self):
        keep = negate(is_error)
        self.assertEqual([keep(r) for r in (TIMEOUT, LOGIN, LAG)], [False, True, True])

    def test_message_contains_ignores_case(self):
        keep = message_contains("timeout")
        self.assertEqual([keep(r) for r in (TIMEOUT, LOGIN, LAG)], [True, False, False])

    def test_filters_combine(self):
        keep = any_of(message_contains("login"), negate(from_db))
        self.assertEqual([keep(r) for r in (TIMEOUT, LOGIN, LAG)], [False, True, False])


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

Hint 1: the idea

Each builder returns a function. Inside it, define a keep(record) that answers yes or no, and return keep without calling it. keep can use the builder’s arguments — predicates, predicate, text — because it closes over them. This lesson’s all_of is the pattern for any_of.

Hint 2: a first step

Write negate first; it’s the shortest — its keep returns the opposite of predicate(record). Once its test passes, write any_of the way Combining filters wrote all_of, with any in place of all.

Show one solution

def any_of(*predicates):
    """Keep records that at least one predicate keeps."""
    def keep(record):
        return any(predicate(record) for predicate in predicates)

    return keep


def negate(predicate):
    """Keep exactly the records predicate doesn't."""
    def keep(record):
        return not predicate(record)

    return keep


def message_contains(text):
    """Keep records whose message contains text, ignoring case."""
    wanted = text.casefold()

    def keep(record):
        return wanted in record.message.casefold()

    return keep

any() of nothing is False, which is why any_of() keeps nothing, and all() of nothing is True, so all_of() keeps everything. Both are the answers that make combining them work without special cases.

message_contains lower-cases text once, when the filter is built, and only the message on each call. casefold() is lower() made for comparisons: it also handles letters like the German ß, which should match “ss”.

What you can now do

Write a factory that returns a configured function, and do its checking and setup once, when it’s built.
Explain what a closure remembers: variables, not values.
Recognise functions made in a loop that all behave like the last one, and fix them with a factory.
Build complex filters out of small, testable parts.

Next: with, and making sure things get cleaned up however a block ends.