Dependency Injection Without a Framework

A domain function that reaches for a global database connection cannot be tested without that database, cannot be given the replica instead of the primary for a report, and cannot be run twice in one process with two different configs. Stagedoor's handlers.py opened its connection at line 12, at import time, and every one of its 3,000 lines used that one object. Passing what a function needs into it, a pool, a clock, a Payrail client, is dependency injection, and the name makes it sound like more than it is. It needs no framework: a constructor, a small Service object built once at startup, and the discipline not to import globals.

Topic 19 drew three layers and let the domain depend on a storage interface. This topic is how the concrete storage gets to the domain without the domain going and fetching it. Stagedoor's startup function is the example, and everything in Chapters 6 through 10 that constructs a pool, a client or a queue consumer does it in that one function.

The Global That Makes Tests Lie

db = connect() at module level runs when the module is imported. Every test that imports the module connects to a database, whether the test needs one or not. Every test shares that one connection and its state, so a test that inserts a hold leaves it for the next test to find. The test that passed alone fails in the suite, and the suite that passed on Monday fails on Tuesday because the order changed. Marek's first suite had 140 tests, took 90 seconds, and had 6 that were marked "flaky" with a retry decorator, which is the polite name for a shared global.

The global is convenient exactly until the second consumer. One handler, one connection, one process: it works, and the code is shorter. The second consumer is the test, or the worker that needs a pool of 5 rather than 20, or the report that should read from pg-replica-a, or the migration script that must not use the app's role. Each of those wants a different connection, and a module-level object has one to give.

Build Once, Pass Down

One function, build_service, takes the typed config of Topic 23 and returns a Service with every process-lifetime dependency attached: the connection pool of Chapter 6, sized 20; the Redis client; the Payrail client of Chapter 10 with its 3-second timeout; the clock; the id generator. The handlers receive the service, through the mechanism of the next section, and call it. They never construct anything. The demonstrated code is that function, and it is the only place in Stagedoor where a pool or a client is created.

class Service:
    seats: SeatRepository
    holds: HoldRepository
    payrail: PayrailClient
    cache: Redis
    clock: Clock
    ids: IdSource

async def build_service(cfg: Config) -> Service:
    pool = AsyncConnectionPool(cfg.database_url, min_size=2, max_size=cfg.pool_size)   # 20
    await pool.open()
    return Service(
        seats=PgSeatRepository(pool),
        holds=PgHoldRepository(pool),
        payrail=PayrailClient(cfg.payrail_url, cfg.payrail_key, timeout_ms=cfg.payrail_timeout_ms),
        cache=Redis.from_url(cfg.redis_url),
        clock=SystemClock(),
        ids=Uuid4Source(),
    )

The function reads top to bottom as an inventory of what the process talks to. The pool is opened once and handed to the two repositories, so both share the same 20 connections. The Payrail client gets its base URL, its key and its timeout from config, so the test suite's build_service can point it at a fake on localhost by changing three values. The clock and the id source look redundant on a first reading, and the next section is why they are not. In the test suite the same function is called with an in-memory SeatRepository swapped in, and every rule test runs against that.

The Clock and the Random Source Are Dependencies

A hold expires 10 minutes after it is created. The test for that rule must create a hold, move time forward 10 minutes, and check that the seat is available again. Code that calls datetime.now() directly can be tested for expiry only by sleeping 600 seconds, which nobody does, so the rule ships untested. With the clock injected, the test's clock is a fake with a set method, and the whole test runs in 2 milliseconds: create at 20:00:00, set 20:09:59, assert held, set 20:10:00, assert available.

The id generator is the same case. Chapter 7's idempotency key and Chapter 3's order public_id both come from uuid4(), and a test that wants to assert "the second request with the same key returns the first response" needs to know the key before the request is made. An IdSource with a fixed sequence in tests and a real UUID generator in production costs one Protocol with one method. Time and randomness are inputs to the rules, and inputs are passed in.

FastAPI's Depends, Used Sparingly

The framework has a mechanism for per-request dependencies: a function declared with Depends runs before the handler, and its return value is passed in as an argument. Stagedoor uses it at the transport edge for exactly three things: to hand the built Service to the handler, to resolve the current user from the token in Chapter 5, and to open a transaction-scoped connection in Chapter 6. The domain never sees Depends, because the domain never sees the framework, which is Topic 19's rule again.

The temptation is to let Depends become the container: a dependency that depends on a dependency that depends on the pool, resolved by the framework on every request, until the handler's signature is nine injected arguments and the construction graph lives in decorators. Everything with a process lifetime is built in build_service and stored once on the app; Depends reads it from there in one line. Per-request things are the only things the framework should construct, and there are three of them.

Interfaces by Protocol

The domain depends on SeatRepository, the storage interface of Topic 19, and it is a Protocol with three methods: get, mark and list_for_event. The storage layer implements it against Postgres. The test suite implements it in memory, with a dictionary. Python's structural typing means neither implementation inherits from anything, registers anywhere or imports the Protocol: a class with those three methods and those signatures is a SeatRepository, and the type checker enforces the contract at the point where the object is passed to build_service.

class SeatRepository(Protocol):
    async def get(self, event_id: int, label: str) -> Seat: ...
    async def mark(self, seat_id: int, status: str, version: int) -> bool: ...
    async def list_for_event(self, event_id: int) -> list[Seat]: ...

class PgSeatRepository:            # storage/seats.py: no base class, no import of the Protocol
    def __init__(self, pool): self.pool = pool
    async def get(self, event_id, label): ...

class MemorySeatRepository:        # tests/fakes.py: a dict of seats, same three methods
    def __init__(self, seats): self.seats = {(s.event_id, s.label): s for s in seats}
    async def get(self, event_id, label): return self.seats[(event_id, label)]

Three method signatures define the whole contract, and two classes meet it without mentioning it. The Postgres one holds a pool and runs SQL; the in-memory one holds a dictionary keyed by event and label. A domain test constructs the second with three seats, builds a service around it, and calls hold_seat; the rule runs unchanged, because the rule was written against the Protocol and never knew which class it was talking to. When Chapter 6 adds the version column to mark, the type checker names every implementation that has not caught up.

Lifetimes

Three lifetimes exist in Stagedoor and nothing lives outside them. The pool, the clients, the clock and the config live for the process: built once, closed at shutdown in Chapter 11. A connection taken from the pool and the request context of Topic 21 live for one request: created by middleware or Depends, released when the response is written. A transaction lives for one unit of work, which Chapter 6 defines as the span from the first statement to the commit, and which is shorter than the request whenever the request also calls Payrail.

Mixing lifetimes is the bug that shares one buyer's transaction with another. A request-scoped connection stored as an attribute on the process-scoped service is visible to the next request on the same instance, and under 3,000 requests a second the next request arrives before the first one's transaction commits. Buyer A's hold is written inside buyer B's transaction; B's rollback undoes A's hold; A's confirmation page says held and the seat map says available. It is a one-line mistake, self.conn = conn in the wrong class, and it is why the service object has no setter for anything with a shorter life than its own.