Function Objects: Solutions

command.py’s commands print a line. A printed line has nothing to undo. The solution therefore swaps in Deposit, a command that changes state, and borrows command_pattern.py’s Macro to hold the list.

If you keep command.py’s Command alias as the type of the macro’s list, the script still prints 15 and 0, but ty reports unresolved-attribute on c.undo(), because an object of type Command has no attribute undo. Command describes one call, so the solution gives the list a Protocol that declares both operations.

# exercise_1.py
from typing import Protocol
from record import record

class UndoableCommand(Protocol):
    def __call__(self) -> None: ...
    def undo(self) -> None: ...

@record
class Deposit:
    account: dict[str, int]
    amount: int

    def __call__(self) -> None:
        self.account["balance"] += self.amount

    def undo(self) -> None:
        self.account["balance"] -= self.amount

class Macro:
    def __init__(self) -> None:
        self.commands: list[UndoableCommand] = []

    def add(self, command: UndoableCommand) -> None:
        self.commands.append(command)

    def run(self) -> None:
        for c in self.commands:
            c()

    def undo_all(self) -> None:
        for c in reversed(self.commands):
            c.undo()

account = {"balance": 0}
macro = Macro()
macro.add(Deposit(account, 10))
macro.add(Deposit(account, 5))
macro.run()
print(account["balance"])
#: 15
macro.undo_all()
print(account["balance"])
#: 0

A bare function is no longer enough. The reason is undo, not state. callable_command.py’s Repeat carries its configuration, and the macro still calls it with (), so a callable handles state. Undo needs more, because a command now answers two requests, __call__() and undo(), while a callable answers one.

Name both requests in one type. A list of undoable commands needs a type with two members, __call__() and undo(). In Python that type is a Protocol, here UndoableCommand. Macro annotates self.commands with UndoableCommand. Deposit matches UndoableCommand with no base class. run() still calls each command with (), as it would call a function.

The GoF Design Patterns shape is a base class with two raise NotImplementedError bodies. A base class makes sense when the commands share implementation. These commands share none.

Remember what to reverse. Deposit must also remember what it did, here the account and the amount, so it can reverse the deposit later. Between calls, a function forgets what each call changed.

Deposit is a record, like Repeat, because its fields, account and amount, keep the values they receive at construction. The record freezes its fields and leaves the dictionary in account mutable, so __call__() and undo() can still update the balance.

Before you move on, name what separates your attempt from this one: a different design, a detail, or an equivalent alternative.

If you annotate chain with a Callable type, as the chapter’s RootFinder is, the script runs and prints the same report, but ty reports unresolved-attribute on both reads of finder.__name__. A Callable describes a call and declares no __name__, so the solution types chain with the Finder protocol, which declares __name__.

# exercise_2.py
from collections.abc import Callable
from typing import Final, Protocol
from record import record

type Fn = Callable[[float], float]

@record
class Failed:
    reason: str

class Finder(Protocol):
    __name__: str
    def __call__(self, f: Fn, a: float,
                 b: float) -> float | Failed: ...

TOLERANCE: Final[float] = 1e-12
MAX_ITER: Final[int] = 200
NO_CONVERGENCE: Final[Failed] = Failed(
    f"no convergence in {MAX_ITER} steps")

def bisection_tol(f: Fn, a: float, b: float,
                  tolerance: float) -> float | Failed:
    if f(a) * f(b) > 0:
        return Failed(f"no root bracketed by [{a}, {b}]")
    while abs(b - a) > tolerance:
        mid = (a + b) / 2
        if f(a) * f(mid) <= 0:
            b = mid
        else:
            a = mid
    return (a + b) / 2

def bisection(f: Fn, a: float, b: float) -> float | Failed:
    return bisection_tol(f, a, b, TOLERANCE)

def secant(f: Fn, a: float, b: float) -> float | Failed:
    x0, x1 = a, b
    for _ in range(MAX_ITER):
        f0, f1 = f(x0), f(x1)
        if f1 == f0:
            return Failed(f"flat step at {x1}")
        x2 = x1 - f1 * (x1 - x0) / (f1 - f0)
        if abs(x2 - x1) < TOLERANCE:
            return x2
        x0, x1 = x1, x2
    return NO_CONVERGENCE

def newton(f: Fn, a: float, b: float) -> float | Failed:
    x = (a + b) / 2
    h = 1e-7
    for _ in range(MAX_ITER):
        slope = (f(x + h) - f(x - h)) / (2 * h)
        if slope == 0:
            return Failed(f"zero slope at {x}")
        step = f(x) / slope
        x -= step
        if abs(step) < TOLERANCE:
            return x
    return NO_CONVERGENCE

def first_root(f: Fn, a: float, b: float,
               chain: list[Finder]) -> float | None:
    for finder in chain:
        match finder(f, a, b):
            case Failed(reason):
                print(f"{finder.__name__}: {reason}")
            case root:
                print(f"{finder.__name__}: {root:.6f}")
                return root
    print("every finder failed")
    return None

def f(x: float) -> float:
    return x * x - 2

first_root(f, 1.0, 1.3, [bisection, secant, newton])
#: bisection: no root bracketed by [1.0, 1.3]
#: secant: 1.414214

def g(x: float) -> float:
    return x * x + 1  # No real root

first_root(g, 0.0, 2.0, [bisection])
#: bisection: no root bracketed by [0.0, 2.0]
#: every finder failed

Give failure a reason. None signals that a handler failed and carries no reason, so a handler that reports why it failed needs a failure value that holds the reason. Failed is that value, a record with one field. Each finder now returns float | Failed.

Report each way to fail. A finder can have more than one way to fail, and each failing return states its own reason. secant() returns a Failed on a flat step, and NO_CONVERGENCE when its loop finishes MAX_ITER iterations. bisection() has one way to fail: it returns a Failed at once when f has the same sign at both ends of the interval.

Tell a root from a failure. first_root() tells a root from a Failed with match. case Failed(reason) prints the reason and lets the loop continue. Any other value matches case root, which prints the root and returns it.

The chapter’s advice about the failure value still holds: first_root() checks the result’s type, not its truthiness. A Failed is never a root, so float | Failed says which result is which.

Name each attempt. Python stores each function’s name in its __name__ attribute, so the report reads finder.__name__ to name the handler that ran, with no extra bookkeeping.

Reading finder.__name__ is why chain needs a Protocol here instead of an alias like the chapter’s RootFinder. ty rejects finder.__name__ on a handler annotated as a Callable, as the opening paragraph shows. Pyright allows it, inferring the attributes of a function object. Finder declares __name__ alongside __call__(), and a function satisfies both.

The listing copies the finders from algorithms.py rather than importing them, because each solution runs on its own.

Before you move on, name what separates your attempt from this one: a different design, a detail, or an equivalent alternative.

If you pass the tolerance positionally, as partial(newton, 0.6), partial() binds 0.6 to f, the first positional parameter. ty reports invalid-argument-type on the 0.6, a float where newton() expects an Fn, and invalid-assignment on the assignment to RootFinder. At run time, calling the finder raises a TypeError when newton() adds the function in a to a float. The solution binds tolerance by keyword, as partial_bisection.py does.

# exercise_4.py
from collections.abc import Callable
from functools import partial
from typing import Final

type Fn = Callable[[float], float]
type RootFinder = Callable[[Fn, float, float], float | None]

MAX_ITER: Final[int] = 200

def newton(f: Fn, a: float, b: float,
           tolerance: float = 1e-12) -> float | None:
    x = (a + b) / 2
    h = 1e-7
    for _ in range(MAX_ITER):
        slope = (f(x + h) - f(x - h)) / (2 * h)
        if slope == 0:
            return None
        step = f(x) / slope
        x -= step
        if abs(step) < tolerance:
            return x
    return None

def newton_within(tolerance: float) -> RootFinder:
    def finder(f: Fn, a: float, b: float) -> float | None:
        return newton(f, a, b, tolerance)
    return finder

def first_root(f: Fn, a: float, b: float,
               chain: list[RootFinder]) -> float | None:
    for finder in chain:
        root = finder(f, a, b)
        if root is not None:
            return root
    return None

def f(x: float) -> float:
    return x * x - 2

coarse_closure = newton_within(0.6)
coarse_partial: RootFinder = partial(newton, tolerance=0.6)
fine_closure = newton_within(1e-12)

for finder in (coarse_closure, coarse_partial,
               fine_closure):
    root = first_root(f, 0.0, 2.0, [finder])
    assert root is not None
    print(f"{root:.6f}")
#: 1.500000
#: 1.500000
#: 1.414214

Keep existing calls working. tolerance becomes a parameter with a default, so every existing call to newton(f, a, b) keeps working.

Configure the finder two ways. The closure and the partial build the same configured strategy in different ways. newton_within() defines a new function whose body supplies the argument. partial(newton, tolerance=0.6) stores the argument and supplies it at the call. Both produce a finder matching RootFinder, so first_root() accepts either with no change.

The two coarse finders print the same wrong-looking answer, 1.500000. That answer shows the tolerance took effect. Newton’s method starts at 1.0 and moves by 0.5 on its first step, to 1.5. A tolerance of 0.6 accepts a step that size, so the loop stops there. The fine finder runs the same code with a tolerance of 1e-12 and matches the true root to six places.

partial is the shorter of the two ways to configure the finder, and it works here because the caller can supply tolerance by keyword. You need a closure when the function has no parameter for the setting. partial() binds arguments. A closure’s body can use the setting in any expression.

Before you move on, name what separates your attempt from this one: a different design, a detail, or an equivalent alternative.

If you index self._handlers[cls] inside the MRO walk, the program prints the same four lines, but each read of a missing class inserts an empty list. After the demo, bus._handlers holds a third key, object, beside the Deposit and BigDeposit keys the solution’s bus holds. The solution reads with .get(cls, []), which returns an empty list and leaves _handlers as it was.

# exercise_5.py
from collections import defaultdict
from collections.abc import Callable
from typing import Any
from record import record

type Handler[E] = Callable[[E], None]

@record
class Deposit:
    amount: int

@record
class BigDeposit(Deposit):
    pass

class EventBus:
    def __init__(self) -> None:
        self._handlers: defaultdict[
            type, list[Handler[Any]]
        ] = defaultdict(list)

    def subscribe[E](self, event_type: type[E],
                     handler: Handler[E]) -> None:
        self._handlers[event_type].append(handler)

    def unsubscribe[E](self, event_type: type[E],
                       handler: Handler[E]) -> None:
        handlers = self._handlers.get(event_type, [])
        if handler in handlers:
            handlers.remove(handler)

    def publish(self, event: object) -> None:
        for cls in type(event).__mro__:  # Parents last
            for handler in self._handlers.get(cls, []):
                handler(event)

def on_deposit(event: Deposit) -> None:
    print(f"deposit {event.amount}")

def on_big(event: BigDeposit) -> None:
    print(f"big deposit {event.amount}")

bus = EventBus()
bus.subscribe(Deposit, on_deposit)
bus.subscribe(BigDeposit, on_big)

bus.publish(BigDeposit(500))
#: big deposit 500
#: deposit 500
bus.publish(Deposit(10))
#: deposit 10

bus.unsubscribe(Deposit, on_deposit)
bus.publish(BigDeposit(500))
#: big deposit 500

Ignore a handler that is not subscribed. unsubscribe() checks if handler in handlers before it calls remove(), since remove() raises a ValueError for a handler not in the list. Whether that case should be silent or loud is a design decision. A silent unsubscribe() matches what publish() does with an event that has no handler. For that event, publish() iterates over empty lists and returns.

Walk the MRO, most specific first. type(event).__mro__ lists the event’s class first and object last, so iterating over it in order calls the most specific handlers first and the inherited ones after. That order meets the “parents last” requirement.

Look up handlers read-only. publish() keeps using .get() for the same reason the chapter gives. Indexing a defaultdict on a read inserts an empty list for every class in every published event’s MRO, object included.

Adding unsubscribe() cannot break an existing caller, since no existing caller calls it. The MRO walk can. A handler subscribed to Deposit starts receiving events of every subclass of Deposit, including subclasses written after the handler. on_deposit now receives every BigDeposit that on_big receives. Delivering subclass events is the intended feature, but it still changes how existing code behaves. Any handler that assumes type(event) is Deposit, or that counts events, now receives more events than before.

Before you move on, name what separates your attempt from this one: a different design, a detail, or an equivalent alternative.

# exercise_6.py
from collections.abc import Callable
from functools import partial

type Command = Callable[[], None]

broken: list[Command] = []
for n in range(3):
    broken.append(lambda: print(n))
for command in broken:
    command()
#: 2
#: 2
#: 2

by_default: list[Command] = []
for n in range(3):
    by_default.append(lambda n=n: print(n))

by_partial: list[Command] = []
for n in range(3):
    by_partial.append(partial(print, n))

def make(n: int) -> Command:
    return lambda: print(n)

by_factory: list[Command] = []
for n in range(3):
    by_factory.append(make(n))

for bound in (by_default, by_partial, by_factory):
    for command in bound:
        command()
#: 0
#: 1
#: 2
#: 0
#: 1
#: 2
#: 0
#: 1
#: 2

# Late lookup, kept on purpose:
settings = {"level": "low"}

def report() -> None:
    print(settings["level"])

settings["level"] = "high"
report()
#: high

Reproduce the late-binding trap. A for loop runs in its enclosing scope, so n is one variable that the loop rebinds three times. All three lambdas close over that one variable rather than over its value. By the time anything calls them, the loop has finished and n holds 2. Nothing is wrong with the lambdas. They read the variable they name, at the moment of the call.

Capture the value at build time. The three fixes all work, and all work the same way. Each one evaluates n while the loop is still running and stores the result. lambda n=n: evaluates the default at definition. partial(print, n) evaluates the argument where it appears. make(n) gives each lambda its own n in its own function scope.

Only make(n) keeps the value private. lambda n=n: exposes the value as a parameter a caller can override. partial(print, n) stores the value in the partial object’s public args attribute, where by_partial[0].args is (0,).

A value computed from the frozen n separates the fixes. The two lambda forms run their body at the call, so lambda n=n: print(n * rate()) and the factory’s lambda both read rate() when the command runs. partial(print, n * rate()) evaluates the product while the loop builds the command.

Read the variable at call time. None of the three preserves late lookup of n. The exercise’s closing question turns on that. If the command must read n when it runs, all three fixes are wrong, because each one freezes the value when the loop builds the command. That command needs the original late lookup, applied to a name that outlives the loop. report() shows that lookup. It reads settings when it runs, so it prints high, the level the script assigns after defining report().

The late-binding trap and late binding as a feature are the same mechanism. Which one you have depends on whether the name you close over still refers to the value you wanted when the command finally runs.

Before you move on, name what separates your attempt from this one: a different design, a detail, or an equivalent alternative.

# exercise_7.py
import inspect
from dataclasses import dataclass
from typing import Final, dataclass_transform
from exceptions import expect

EVENTS: Final[set[type]] = set()

@dataclass_transform(frozen_default=True)
def event[E](cls: type[E]) -> type[E]:
    built = dataclass(frozen=True, slots=True)(cls)
    EVENTS.add(cls)  # The change: cls, not built
    return built

@dataclass_transform(frozen_default=True)
def handler[H](cls: type[H]) -> type[H]:
    call = vars(cls)["__call__"]
    sig = inspect.signature(call)
    handled = list(sig.parameters.values())[1].annotation
    if handled not in EVENTS:
        raise TypeError(f"{cls.__name__}: not an @event")
    return dataclass(frozen=True, slots=True)(cls)

@event
class Deposit:
    amount: int

print(len(EVENTS), Deposit in EVENTS)
#: 1 False
print(type(Deposit(5)) in EVENTS)
#: False

class Announce:
    prefix: str
    def __call__(self, event: Deposit) -> None:
        print(f"{self.prefix} deposit {event.amount}")

expect(TypeError, handler, Announce)
#: [TypeError] Announce: not an @event

The listing copies event() and handler() with the change made, trims them to the lines the question reaches, and applies handler() as a call so that expect() can report the failure.

Register the wrong class. The import runs cleanly through the three @event classes in bank_events.py. That clean run hides the mistake. EVENTS holds one class for each, and each is the wrong one. dataclass() with slots=True builds a new class, event() returns that new class, and the class statement binds Deposit to what event() returns. The set holds the original class. No name references that class, and no event is an instance of it.

Refuse the first handler. The import stops at the first @handler, on Announce. handler() reads the annotation on event and gets Deposit, the class dataclass() returned. EVENTS holds the original class instead, so handler() raises TypeError: Announce: not an @event, although Deposit went through @event a few lines earlier. If something else created the handlers, publish() would raise a TypeError on every event, since type(event) is the returned class too.

Before you move on, name what separates your attempt from this one: a different design, a detail, or an equivalent alternative.