Chapter 25 · Solutions

Surrogate: Solutions

1. A virtual proxy that answers the cheap requests itself

Extend virtual_proxy.py’s Lazy so it answers one attribute itself, a description string given at construction, without building Expensive. Count the accesses it answers that way, and report the count when Lazy builds Expensive. Confirm that reading description several times builds nothing, and that the first query() reports the count.

Where to look

Virtual Proxy builds Expensive inside __getattr__(), which runs only when normal lookup fails. Give Lazy a property for description, so Python finds it on the class and skips the fallback. Increment a counter in that property, and print the counter at the moment the fallback builds the real object.

The shape

# The shape of exercise_1.py
from typing import Any

class Expensive:
    def __init__(self) -> None:
        ...

    def query(self) -> str:
        ...

class Lazy:
    def __init__(self, description: str) -> None:
        ...

    @property
    def description(self) -> str:
        ...

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

Solution

If you store description as an ordinary instance attribute, normal lookup finds it, so the three reads build nothing. Each read returns the stored string, though, so the counter stays at zero. The first query() then reports 0 answered before build. A property also stops the lookup at the class, before __getattr__(), and it runs a method on each read, so the solution counts inside that method.

# 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())  # [1]
#: 3 answered before build
#: Expensive built
#: result
print(p.query())  # [2]
#: result

Answer description without building. description is a property of Lazy, so normal lookup finds it on the class and stops there. The three reads build nothing, and each one increments _answered.

Build the real object on demand. At [1], query is the first name missing from the proxy, so __getattr__() runs, reports the count, and builds the real object. At [2], __getattr__() runs again, finds _real set, and forwards the lookup with no report and no build.

The counter records how much work the proxy saved. The proxy serves three requests from a string it holds from the start and defers the slow construction until the first query(). GoF’s image proxy is the same design. It answers an image’s size from stored numbers and loads the pixels when something first draws the image.

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

2. A per-method tally in the counting proxy

Change CountingProxy in counting_proxy.py to keep a per-method tally in a collections.Counter instead of a single total. Confirm the tally reports f called twice and g called once.

Where to look

Smart Reference wraps each forwarded call to count it. __getattr__() receives the attribute name, so use the name as the key into a collections.Counter in place of the single total. Increment the entry inside the wrapper, before forwarding the call.

The shape

# The shape of exercise_2.py
from collections import Counter
from typing import Any

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

class CountingProxy:
    def __init__(self, impl: Any) -> None:
        ...

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

Solution

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. That version counts lookups rather than calls, so 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 in an integer calls, this one keeps a Counter keyed by method name. __getattr__() receives the name of the attribute, so the wrapper adds one to that name’s entry before forwarding the call. The final print() shows f called twice and g once.

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

3. A simple copy-on-write list

Create a simple copy-on-write list. Its share() returns a second list over the same data and increments a reference count. The first append() through a shared list copies the data before changing it. Confirm that the two lists share their data before the write and hold separate copies after it.

Where to look

Smart Reference shows a surrogate doing extra work around each use of an implementation. Keep the data and a count of owners together in one small shared object, and let share() hand out the same object after adding one to the count. Have append() check the count and, when more than one owner exists, copy the data into a new object first.

The shape

# The shape of 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:
        ...

    def share(self) -> CowList:
        ...

    def append(self, item: object) -> None:
        ...

    def __len__(self) -> int:
        ...

    def __repr__(self) -> str:
        ...

Solution

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 and leaves the data shared, 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, and through it one list. owners counts the CowLists that point at that Box. share() copies a reference to the Box and increments owners.

Copy before a shared write. append() copies the data only when owners > 1. b.append(4) decrements the shared Box’s count, since b stops referring to that Box. Then append() points b at a private Box holding a fresh copy of the data and appends to that private copy. a keeps the original Box, whose list still holds [1, 2, 3].

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

4. Why the typo reports as `RecursionError`

In counting_proxy.py, misspell self._impl as self._imp inside __getattr__() and run counting_demo.py. Use the fallback behavior this chapter describes to explain why the failure reports as RecursionError rather than an AttributeError naming the typo.

Where to look

The Recursion Trap and Forwarding with __getattr__() describe a method that Python calls only after normal lookup fails. Trace what Python does when the first line inside that method reads a missing name. Use expected() from exceptions to catch the failure in the listing.

The shape

# The shape of exercise_4.py
from typing import Any
from exceptions import expected

class Implementation:
    def f(self) -> None: ...

class BrokenProxy:
    def __init__(self, impl: Any) -> None:
        ...

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

Solution

# 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, another missing name, so Python calls __getattr__("_imp"), which starts by reading self._imp. Each call to __getattr__() starts another missing-attribute lookup, so the calls nest until they exceed Python’s recursion limit. Python then raises a RecursionError before any AttributeError can name the typo.

The trap is specific to the fallback method. __getattr__() runs only when normal lookup fails, so any missing name it reads makes Python call __getattr__() again.

The chapter’s working version is safe because __init__() assigns self._impl, so normal lookup finds _impl on the instance and returns it. A proxy built through object.__new__() skips __init__(), so it fails the same way on its first lookup that falls through to __getattr__().

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

5. A connection pool that hands out proxies

Create a program similar to a DBMS that allows only a set number of connections at a time. Implement this with a system modeled on Singleton that controls the number of “connection” objects it creates. When a user finishes with a connection, the system must check that connection back in for reuse. To guarantee this, return a proxy instead of a reference to the actual connection, and design the proxy to release the connection back to the system.

Where to look

Protection Proxy shows a surrogate that controls access to an implementation. Let only a Pool class create the Connection objects, and have acquire() return a proxy that forwards through __getattr__(). Make the proxy a context manager whose __exit__() returns the connection to the pool and drops its own reference.

The shape

# The shape of exercise_5.py
from typing import Any, Final, Self
from exceptions import expect, expected

POOL_SIZE: Final[int] = 2

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

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

    def query(self, sql: str) -> str:
        ...

class Pool:
    def __init__(self, size: int) -> None:
        ...

    def available(self) -> int:
        ...

    def acquire(self) -> ConnectionProxy:
        ...

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

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

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

    def __enter__(self) -> Self:
        ...

    def __exit__(self, *exception: object) -> None:
        ...

Solution

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() still answers through connection 0. When a later acquire() lends that connection 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, expected

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
with expected(RuntimeError):
    c1.query("select 3")
#: [RuntimeError] connection already released

Limit who creates connections. Pool alone creates Connection objects, all of them in its constructor. Pool controls creation as a Singleton class does, with the limit set at POOL_SIZE objects instead of one.

Hand out a stand-in. acquire() hands the client a ConnectionProxy in place of a Connection. The proxy forwards query() through __getattr__() and adds one action of its own: returning that connection to the pool. Singleton’s exercise 2 builds the same pool without a proxy, so each caller there must remember to call release(). Here the proxy takes over that step.

Return the connection on exit. The proxy is also a context manager. __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, which the pool may lend to someone else. The check in __getattr__() then raises a RuntimeError on any later use of the proxy, so each connection serves one client at a time. The call to c1.query() after both blocks shows that error.

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.

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

6. Forwarding `len()` explicitly

len(p) raises a TypeError on the Proxy that dunder_bypass.py imports from proxy_getattr.py. Give that Proxy a __len__() that forwards to the implementation, and confirm len(p) returns 2. Then explain why __getattr__() could not have supplied it.

Where to look

Special Methods Bypass __getattr__() explains why len(p) fails on a proxy that forwards through __getattr__() alone. Define __len__() on the proxy class and have it call len() on the implementation. For the explanation, consider where len() looks for __len__().

The shape

# The shape of exercise_6.py
from typing import Any

class Words:
    def __init__(self) -> None:
        ...

    def __len__(self) -> int:
        ...

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

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

    def __len__(self) -> int:
        ...

Solution

If Proxy forwards through __getattr__() alone, p.__len__() returns 2, but len(p) raises a TypeError: object of type 'Proxy' has no len(). The type checker reports the call as invalid-argument-type, since Proxy does not match Sized. The solution defines __len__() on the class, where len() looks for it.

# 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() looks up __len__() on type(p) and calls what it finds there. That lookup bypasses the instance, and with it __getattr__(), which Python calls after a failed lookup on the instance. Python’s operators and built-in functions look up an 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. len() reads better.

To forward many dunders, you write one forwarding method per dunder, or generate the methods in a loop over a list of names and assign them onto the class.

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

7. A `change_to()` that refuses a narrower implementation

Extend Surrogate in state_surrogate.py so change_to() rejects an implementation missing a method the current one has, and explain why the type checker could not have reported that swap.

Where to look

State swaps the implementation behind a surrogate with change_to(). Build a set of public callable names with dir() and getattr() for the current implementation and for the new one. Subtract the new set from the current one and raise a TypeError when anything remains. For the explanation, consider what type the surrogate gives its implementation.

The shape

# The shape of exercise_7.py
from typing import Any
from exceptions import expect

def methods(obj: object) -> set[str]:
    ...

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

    def change_to(self, new: Any) -> None:
        ...

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

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

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

Solution

# 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 new. The checker sees Any for both. Any is the choice state_surrogate.py makes so that one surrogate accepts any implementation.

Annotating the stored implementation and new against a Protocol gives a different guarantee: one shape that every implementation must meet. A Protocol cannot express “at least what the last implementation had,” because that comparison relates two runtime values rather than two declarations.

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

Surrogate: Solutions

1. A virtual proxy that answers the cheap requests itself

2. A per-method tally in the counting proxy

3. A simple copy-on-write list

4. Why the typo reports as RecursionError

5. A connection pool that hands out proxies

6. Forwarding __len__() explicitly

7. A change_to() that refuses a narrower implementation

4. Why the typo reports as `RecursionError`

6. Forwarding `len()` explicitly

7. A `change_to()` that refuses a narrower implementation