Surrogate: Solutions

If you store description as an ordinary instance attribute, normal lookup finds it and the three reads build nothing, but no code runs on a read. The counter stays at zero, so the first query() reports 0 answered before build. A property runs a method on each read, so the solution counts there without reaching __getattr__().

# exercise_1.py
from typing import Any

class Expensive:
    def __init__(self) -> None:
        print("Expensive built")

    def query(self) -> str:
        return "result"

class Lazy:
    def __init__(self, description: str) -> None:
        self._description = description
        self._answered = 0
        self._real: Expensive | None = None

    @property
    def description(self) -> str:
        self._answered += 1
        return self._description

    def __getattr__(self, name: str) -> Any:
        if self._real is None:
            print(f"{self._answered} answered before build")
            self._real = Expensive()
        return getattr(self._real, name)

p = Lazy("a slow query")
for _ in range(3):
    print(p.description)
#: a slow query
#: a slow query
#: a slow query
print(p.query())
#: 3 answered before build
#: Expensive built
#: result
print(p.query())
#: result

Answer the cheap request without building. description is a property on the proxy, so Python finds it without calling __getattr__(), and the three reads build nothing. Each one increments _answered.

Build the real object on demand. The first query() is the first name the proxy lacks, so __getattr__() runs, reports the count, and builds the real object; the second query() finds _real set and forwards without reporting or building.

The counter records how much work the proxy saved: three requests served from a string the proxy held from the start, with the slow construction pushed past all of them. GoF’s image proxy is the same design, answering an image’s size from stored numbers while the pixels stay unloaded until something draws them.

If you increment self.calls[name] in __getattr__() before the callable() test, the demo still prints 2 1, because each lookup there leads to one call. A lookup without a call counts too: evaluating p.f is p.f adds two to f’s tally. The solution counts inside counted, so the tally advances at the call, as it does in the chapter’s CountingProxy.

# exercise_2.py
from collections import Counter
from typing import Any

class Implementation:
    def f(self) -> None: print("f()")
    def g(self) -> None: print("g()")

class CountingProxy:
    def __init__(self, impl: Any) -> None:
        self._impl = impl
        self.calls: Counter[str] = Counter()

    def __getattr__(self, name: str) -> Any:
        attr = getattr(self._impl, name)
        if callable(attr):
            def counted(*args: Any, **kwargs: Any) -> Any:
                self.calls[name] += 1
                return attr(*args, **kwargs)
            return counted
        return attr

p = CountingProxy(Implementation())
p.f()
#: f()
p.g()
#: g()
p.f()
#: f()
print(p.calls["f"], p.calls["g"])
#: 2 1

Tally each call by name. Where the chapter’s CountingProxy keeps one total, this one tallies per method name. __getattr__() receives the name of the attribute, so the wrapper charges the count to that name before forwarding. The single calls integer becomes a Counter. The final print() shows f called twice and g once.

If you build the private Box around self._box.data without the list(...) copy, b gets a Box of its own that holds the same list. b.append(4) then changes a too: the demo prints [1, 2, 3, 4] [1, 2, 3, 4] while a._box is b._box reports False. A new Box gives b its own owner count but not its own data, so the solution copies the list before the write.

# exercise_3.py
from collections.abc import Sequence
from dataclasses import dataclass

@dataclass
class Box:
    data: list[object]
    owners: int = 1

class CowList:
    def __init__(self, data: Sequence[object] | None = None,
                 _box: Box | None = None) -> None:
        self._box = (
            _box if _box is not None
            else Box(list(data or [])))

    def share(self) -> CowList:
        self._box.owners += 1
        # Shares the same Box, for now
        return CowList(_box=self._box)

    def append(self, item: object) -> None:
        if self._box.owners > 1:
            # Shared Box: copy before mutating
            self._box.owners -= 1
            self._box = Box(list(self._box.data))
        self._box.data.append(item)

    def __len__(self) -> int:
        return len(self._box.data)

    def __repr__(self) -> str:
        return repr(self._box.data)

a = CowList([1, 2, 3])
b = a.share()
print(a._box is b._box, a._box.owners)
#: True 2
b.append(4)
print(a, b)
#: [1, 2, 3] [1, 2, 3, 4]
print(a._box is b._box)
#: False

Share the data and count its owners. a and b start out sharing one Box, the same underlying list, with owners tracking how many CowLists point at that Box. share() costs almost nothing: it copies a reference and bumps a count.

Copy before a shared write. append() copies the data, and only when owners > 1. b.append(4) detaches b into its own private Box holding a fresh copy of the data, decrements the shared Box’s count (since b is no longer one of its owners), then appends to that private copy. Since no one called a.append(), a still points at the original, untouched Box. The first write triggers the copy, and only the list that writes pays for it.

# exercise_4.py
from typing import Any
from exceptions import expected

class Implementation:
    def f(self) -> None: print("f()")

class BrokenProxy:
    def __init__(self, impl: Any) -> None:
        self._impl = impl
        self.calls = 0

    def __getattr__(self, name: str) -> Any:
        attr = getattr(self._imp, name)  # Deliberate typo
        if callable(attr):
            def counted(*args: Any, **kwargs: Any) -> Any:
                self.calls += 1
                return attr(*args, **kwargs)
            return counted
        return attr

p = BrokenProxy(Implementation())
with expected(RecursionError):
    p.f()
#: [RecursionError] maximum recursion depth exceeded

Python finds no f on the instance or on BrokenProxy, so it calls __getattr__("f"). That call starts by reading self._imp, which does not exist either, so Python calls __getattr__("_imp"), which starts by reading self._imp. Each attempt to report the missing attribute creates another missing-attribute lookup, and the stack runs out before Python can raise an AttributeError.

The trap is specific to the fallback method. __getattr__() runs only when normal lookup fails, so any missing name it touches sends Python straight back into __getattr__(). Reading self._impl, which __init__() did assign, resolves normally without reaching __getattr__(). That normal lookup is why the chapter’s working version is safe and BrokenProxy is not. A proxy whose __init__() did not run (an instance built through object.__new__(), for example) fails the same way on its first attribute access.

If you leave out the line in __exit__() that sets _connection to None, the pool gets each connection back, but the proxy keeps its reference. After both blocks end, c1.query() answers through connection 0, and when a later acquire() lends connection 0 to another client, both proxies query through it. The solution clears the reference, so a released proxy raises a RuntimeError instead.

# exercise_5.py
from typing import Any, Final, Self
from exceptions import expect

POOL_SIZE: Final[int] = 2

class PoolExhausted(RuntimeError):
    "No connection is free."

class Connection:
    def __init__(self, number: int) -> None:
        self.number = number

    def query(self, sql: str) -> str:
        return f"connection {self.number}: {sql}"

class Pool:
    def __init__(self, size: int) -> None:
        self._size = size
        self._free = [Connection(n) for n in range(size)]

    def available(self) -> int:
        return len(self._free)

    def acquire(self) -> ConnectionProxy:
        if not self._free:
            raise PoolExhausted(f"all {self._size} in use")
        return ConnectionProxy(self, self._free.pop(0))

    def release(self, connection: Connection) -> None:
        self._free.append(connection)

class ConnectionProxy:
    def __init__(self, pool: Pool,
                 connection: Connection) -> None:
        self._pool = pool
        self._connection: Connection | None = connection

    def __getattr__(self, name: str) -> Any:
        if self._connection is None:
            raise RuntimeError(
                "connection already released")
        return getattr(self._connection, name)

    def __enter__(self) -> Self:
        return self

    def __exit__(self, *exception: object) -> None:
        if self._connection is not None:
            self._pool.release(self._connection)
            self._connection = None

pool = Pool(POOL_SIZE)
with pool.acquire() as c1:
    print(c1.query("select 1"))
    with pool.acquire() as c2:
        print(c2.query("select 2"))
        print("free:", pool.available())
        expect(PoolExhausted, pool.acquire)
    print("inner released:", pool.available())
#: connection 0: select 1
#: connection 1: select 2
#: free: 0
#: [PoolExhausted] all 2 in use
#: inner released: 1
print("outer released:", pool.available())
#: outer released: 2

Limit who creates connections. Pool builds every Connection in its constructor, and nothing else creates one. Pool controls creation as a Singleton class does, with the limit raised from one object to POOL_SIZE.

Hand out a stand-in. The client holds no Connection. acquire() hands back a ConnectionProxy, which forwards query() through __getattr__() and owns the one job the connection cannot do for itself: returning that connection to the pool.

Return the connection on exit. The proxy is also a context manager (Context Managers). __exit__() runs whether the block ends normally or raises an exception, so “must check that connection back in” becomes a guarantee.

Refuse use after release. __exit__() also drops the proxy’s reference to the connection, so a released proxy cannot keep using a connection that now belongs to someone else. The check in __getattr__() reports that misuse instead of letting two clients share one connection. ConnectionProxy is a protection proxy and a smart reference at once: it controls access, and it adds an action (the check-in) around each loan of the connection.

# exercise_6.py
from typing import Any

class Words:
    def __init__(self) -> None:
        self.items = ["spam", "eggs"]

    def __len__(self) -> int:
        return len(self.items)

class Proxy:
    def __init__(self, impl: Any) -> None:
        self.__implementation = impl

    def __getattr__(self, name: str) -> Any:
        return getattr(self.__implementation, name)

    def __len__(self) -> int:
        return len(self.__implementation)

p = Proxy(Words())
print(len(p))
#: 2

__getattr__() could not have supplied __len__() because len() does not look the name up on the instance. len() asks type(p) for __len__() and calls what it finds there. That lookup skips the instance, so no instance lookup fails, and a failed instance lookup is the one event that calls __getattr__(). Python looks up every implicitly invoked special method this way, so the method must exist on the proxy’s class.

Forward the special method explicitly. __len__() here delegates with len(self.__implementation) rather than self.__implementation.__len__(). Both give the same answer, and len() reads better. To forward many dunders, you write one such method per dunder, or generate them in a loop over a list of names and assign them onto the class.

# exercise_7.py
from typing import Any
from exceptions import expect

def methods(obj: object) -> set[str]:
    return {
        name
        for name in dir(obj)
        if not name.startswith("_")
        and callable(getattr(obj, name))
    }

class Surrogate:
    def __init__(self, implementation: Any) -> None:
        self.__implementation = implementation

    def change_to(self, new: Any) -> None:
        missing = (methods(self.__implementation)
                   - methods(new))
        if missing:
            raise TypeError(f"missing: {sorted(missing)}")
        self.__implementation = new

    def __getattr__(self, name: str) -> Any:
        return getattr(self.__implementation, name)

class Full:
    def f(self) -> None: print("Full.f()")
    def g(self) -> None: print("Full.g()")

class Lacking:
    def f(self) -> None: print("Lacking.f()")

s = Surrogate(Full())
s.f()
#: Full.f()
expect(TypeError, s.change_to, Lacking())
#: [TypeError] missing: ['g']
s.g()  # The old implementation is still in place
#: Full.g()

List the reachable methods. methods() reports the public callables an object carries, the set a caller can reach through the surrogate’s __getattr__().

Refuse a narrower replacement. change_to() compares the two sets and refuses the swap when the replacement drops a name the current implementation answers. The surrogate keeps its current implementation, so s.g() still works after the rejected swap.

The type checker cannot make this decision. The decision compares the type of the implementation the surrogate holds right now with the type of the argument, and the checker knows neither: both are Any, because __getattr__() delegation deliberately leaves the implementation’s type untracked. Annotating both against a Protocol states a fixed shape that every implementation must meet, a different guarantee. A Protocol cannot express “at least what the last implementation had,” because that comparison relates two runtime values rather than two declarations.