Effect Management: Solutions

# exercise_1.py
from typing import Protocol

class Ask(Protocol):
    def ask(self, prompt: str) -> str: ...

class Tell(Protocol):
    def tell(self, message: str) -> None: ...

def greet(ask: Ask, tell: Tell) -> None:
    name = ask.ask("What is your name? ")
    tell.tell(f"Hello, {name}!")

class Console:
    def ask(self, prompt: str) -> str:
        return input(prompt)

    def tell(self, message: str) -> None:
        print(message)

class Scripted:
    def ask(self, prompt: str) -> str:
        return "Alice"

greet(Scripted(), Console())  # Real tell, scripted ask
#: Hello, Alice!

An interactive run of greet(Console(), Console()) looks like this:

What is your name? Alice
Hello, Alice!

exercise_1.py binds the real Console for tell and keeps the scripted ask, because a book listing that calls input() runs with no terminal attached. The substitution is the point either way: one Console instance satisfies both protocols, so you can pass the same object for both parameters. You can also replace either parameter with a double, and the other one is unaffected.

Declare what the function needs. greet() requires no change, and could not have required one. It names two capabilities it needs and calls methods on them. Its body names none of Console, input(), print(), or Scripted, so changing the binding affects nothing there. That is the delayed-binding payoff: the choice of implementation moves to the call site, where a test can choose differently from production.

Satisfy the protocols structurally. The type checker still enforces the protocols after the choice moves. Console inherits from nothing and declares no relationship to Ask or Tell, but it has the two methods with the right signatures, so it satisfies both protocols structurally. If you give Console a tell() that returns a str, ty reports invalid-argument-type at the greet(...) call, not at the class definition, because the class does not declare which protocol it means to satisfy.

# exercise_2.py
from dataclasses import dataclass, field
from typing import Protocol

class Ask(Protocol):
    def ask(self, prompt: str) -> str: ...

class Tell(Protocol):
    def tell(self, message: str) -> None: ...

class Log(Protocol):
    def log(self, message: str) -> None: ...

def format_greeting(name: str, log: Log) -> str:
    log.log(f"formatting greeting for {name}")
    return f"Hello, {name}!"

def greet(ask: Ask, tell: Tell, log: Log) -> None:
    log.log("greet started")
    name = ask.ask("What is your name? ")
    tell.tell(format_greeting(name, log))

def session(ask: Ask, tell: Tell, log: Log) -> None:
    greet(ask, tell, log)

def menu(ask: Ask, tell: Tell, log: Log) -> None:
    session(ask, tell, log)

def main(ask: Ask, tell: Tell, log: Log) -> None:
    menu(ask, tell, log)

class Scripted:
    def ask(self, prompt: str) -> str:
        return "Alice"

@dataclass
class Capture:
    messages: list[str] = field(default_factory=list)

    def tell(self, message: str) -> None:
        self.messages.append(message)

    def log(self, message: str) -> None:
        self.messages.append(f"LOG: {message}")

captured = Capture()
main(Scripted(), captured, captured)
for line in captured.messages:
    print(line)
#: LOG: greet started
#: LOG: formatting greeting for Alice
#: Hello, Alice!

Five signatures name the new Effect, and only two of them use it.

Log at the point of use. One of the five is new rather than edited: format_greeting(), the helper that uses the Log. You must edit four existing signatures. greet() both uses a Log and accepts one, to hand down to format_greeting().

Hand the Effect down the call chain. session(), menu(), and main() each gain a log parameter that they hand to the next function. Three of the five name a Log they do not use. Those functions sit between the Effect’s user and the call site that binds it, and they pay for an Effect whose sole use in their bodies is forwarding it. Their signatures now describe a capability they do not exercise, so a reader of menu() learns something false about what menu() does.

The cost also scales the wrong way. Adding a fourth Effect later means walking the same chain again, and the chain is longer in a real program than in this one. The alternative most codebases pick, a module-level logger, removes the parameter by removing the choice: the function no longer says it logs, and a test can no longer bind the logger differently.

An Effect Management System collapses the parameter lists and the call sites, not the signatures. format_greeting() declares in its return type that it needs a Log, and so does every function on the path to it, but none of them gains an unused parameter, no call site changes, and the type checker names each declaration you miss. You still supply the binding, but at one place near the top, where the program decides what a Log means. Stateless shows that shape with a real library: in its audit_log.py, greet_logged() and its caller greet_all() both carry a Need[Log] in the Depend return type, while greet() stays unchanged.

If you keep NonZero’s self.value == 0 test under the new name, PositiveInt(-1) constructs without complaint, and slope(10, PositiveInt(-1)) returns -10.0. The demo’s expect() loop then raises an AssertionError (“no exception raised”) on -1, and the type checker reports nothing, because every argument has the right type. The solution’s __post_init__() tests value <= 0 instead, which rejects both bad values.

# exercise_4.py
from exceptions import expect
from record import record

@record
class PositiveInt:
    value: int

    def __post_init__(self) -> None:
        if self.value <= 0:
            raise ValueError(
                f"PositiveInt needs a positive value: "
                f"{self.value}")

def slope(rise: int, run: PositiveInt) -> float:
    return rise / run.value

print(slope(10, PositiveInt(2)))
#: 5.0
for bad in (0, -1):
    expect(ValueError, PositiveInt, bad)
#: [ValueError] PositiveInt needs a positive value: 0
#: [ValueError] PositiveInt needs a positive value: -1

Cover both bad values with one predicate. The original slope_catch.py splits the guarding in a way that is easy to miss. validate() rejects negatives but lets zero through, and the try catches zero but says nothing about negatives. Two mechanisms in two places cover the two bad values, and neither one tells you the other exists. NonZero inherits half of that split. One predicate, value <= 0, covers both, because “positive” is a single idea and “not zero, and also not negative” is the same idea described as two exceptions.

Trust the parameter’s type. Both checks disappear from slope(), and so does everything they brought with them. A PositiveInt cannot hold zero, so the try/except ZeroDivisionError goes. It cannot hold a negative either, so the call to validate() goes, and validate() along with it. Only the division remains.

Fail at the construction site. The cost moves rather than vanishing. PositiveInt(bad) still raises an exception, at the boundary where an untrusted number enters the program, and a caller reading from a file or a form must still handle it. The count changes: one construction site instead of every function that touches the value. Every function downstream of a PositiveInt is pure with respect to this failure, and none of them spends a line of code on it. Each signature says which values the function accepts, instead of leaving that to a docstring.

If you leave out the await in the list comprehension, sum() receives a list of coroutines rather than prices. ty reports no-matching-overload at the sum() call, and the asyncio.run() call raises a TypeError for adding an int and a coroutine, with a RuntimeWarning that price_of_async was never awaited. Calling an async function builds a coroutine and runs none of its body, so the solution awaits each call to get its float.

# exercise_5.py
import asyncio

PRICES = {"apple": 1.5, "pear": 2.0}

def price_of(item: str) -> float:
    return PRICES[item]

def total_price(items: list[str]) -> float:
    return sum(price_of(item) for item in items)

async def price_of_async(item: str) -> float:
    await asyncio.sleep(0)
    return PRICES[item]

async def total_price_async(items: list[str]) -> float:
    return sum(
        [await price_of_async(item) for item in items])

basket = ["apple", "pear"]
print(total_price(basket))
#: 3.5
print(asyncio.run(total_price_async(basket)))
#: 3.5
description = price_of_async("apple")
print(type(description).__name__)
#: coroutine
description.close()  # Never awaited, so close it explicitly

Making the helper async forces four changes, and none of them is optional.

Await the helper from a coroutine. price_of_async("apple") now returns a coroutine instead of a float, as the last print() shows, so total_price() cannot sum the results until each call has an await. Only an async def may contain await, so total_price() becomes total_price_async().

Collect the results before summing. The argument to sum() gains brackets. A generator expression with an await inside it is an asynchronous generator, which sum() cannot iterate: ty reports no-matching-overload, and the call raises a TypeError. The list comprehension awaits each price and hands sum() a list.

Run the coroutine at the edge. The callers of total_price_async() then must become async in turn, and the propagation stops only at asyncio.run(), the boundary that discharges the Effect.

That propagation is Effect tracking. The Effect appears in the type: ty reports price_of_async’s return as CoroutineType[Any, Any, float], not float. A caller that forgets await then gets a type error rather than a mysterious value. The Effect travels outward one caller at a time, as the chapter says an Effect propagates, and you cannot leave a caller out. It reaches the edge of the program, where a single call runs it. async satisfies property 1 of the three-item list without anyone calling it an Effect system.

It satisfies neither of the other two.

It does not separate the interface from the implementation. await price_of_async(item) names no capability. It says “run this particular coroutine,” and the coroutine’s body decides what awaiting means. Compare Ask in ask_tell.py, where greet() names the capability and stays silent about the implementation’s source. async has no equivalent of writing a function against “something awaitable that yields a price” and choosing the implementation later.

It does not bind the implementation later. asyncio.run() chooses an event loop, and that choice sounds like late binding until you ask what it lets you swap. Choosing a loop does not let a test substitute a different meaning for the awaits inside: you settle what those awaits mean when you write the coroutine. A test that needs fake prices must still inject price_of_async, by the same hand-threading this chapter’s exercise 2 measures. The event loop is a scheduler, not a handler.

So async is an Effect-tracking system rather than a full EMS, in the same sense as most of the AI languages in Custom AI Languages with Effects. It tracks one fixed Effect, chosen by the language, with the implementation set at the call site. That is also why the propagation feels like a nuisance rather than a benefit: you get the bookkeeping cost of Effect tracking without the delayed binding that would repay it.