Skip to content

Composition Root

Summary

All wiring happens in exactly one module. It builds a flat resource registry and exposes it to the HTTP layer and to scripts.

The resource registry

init.py returns a plain dict[str, Any]. Keys follow two conventions:

  • <resource>_repo — a repository adapter for a resource.
  • <resource>_svc — a service for a resource.

Plus named infrastructure adaptors:

  • http_cache_store — response cache backend.
  • progress_bus — progress/event bus.
  • <vendor>_<kind> — fetchers, clients, stores.

The API stores the dict on app.state.db_resources during lifespan startup. Handlers read from it; nothing imports concrete adapters.

Shape

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
_resources: dict = {}
_initialized = False

async def init(settings=None) -> dict:
    global _initialized
    if _initialized:
        return _resources

    await setup_database(uri=settings.mongo_uri, db_name=settings.mongo_db)

    # 1. leaf repositories
    user_repo = create_user_repo()

    # 2. dependent repositories (dependencies passed by keyword)
    wallet_repo = create_wallet_repo(user_repo=user_repo)

    # 3. services (build via the DI library, injecting the repo)
    _resources["user_repo"]   = user_repo
    _resources["wallet_repo"] = wallet_repo
    _resources["user_svc"]    = get_user_service(user_repo)
    _resources["wallet_svc"]  = get_wallet_service(wallet_repo)

    # 4. infrastructure adaptors (optional)
    _resources["http_cache_store"] = HttpCacheStore(...)
    _resources["progress_bus"]     = ProgressEventBus()

    _initialized = True
    return _resources

async def shutdown() -> None:
    global _initialized
    await close_stores(_resources)
    await shutdown_database()
    _resources.clear()
    _initialized = False

Typed accessors

Expose a get_svc(name) helper with one @overload per key so scripts get typing without importing concrete classes:

1
2
3
4
5
6
7
@overload
def get_svc(name: Literal["user_svc"]) -> UserService: ...
@overload
def get_svc(name: Literal["wallet_svc"]) -> WalletService: ...

def get_svc(name: str):
    return _resources[name]

Also expose get_db() (raw database handle) and pure helpers such as compute_fingerprint(payload) (canonical hash used for deduplication).

Rules

  • init() is idempotent.
  • The composition root is the only place that imports concrete adapters.
  • Handlers never call init(); they read request.app.state.db_resources.
  • Scripts call init() / shutdown() around their work.