Skip to content

Controlling the composition root

In dependency injection technical jargon, the "composition root" is a single logical place (function, module, etc.) where all of your dependendencies are "composed" together and abstractions are bound to concrete implementations.

You can acheive this in Xpresso if your are willing to take some control of application initialization.

In many cases, this will let you cut out intermediary dependencies (e.g. a dependency to get a database connection or load a config from the environment): you can load your config, create your database connection and bind your repos/DAOs so that your application never has to know about a config or even what database backend it is using.

import sqlite3
from dataclasses import dataclass
from typing import List

from xpresso import App, FromJson, Path

class SupportsWordsRepo:
    def add_word(self, word: str) -> None:
        raise NotImplementedError

class SQLiteWordsRepo(SupportsWordsRepo):
    conn: sqlite3.Connection

    def add_word(self, word: str) -> None:
        with self.conn:
            self.conn.execute("SELECT ?", (word,))

def add_word(repo: SupportsWordsRepo, word: FromJson[str]) -> str:
    return word

routes = [Path("/words/", post=add_word)]

def create_app() -> App:
    conn = sqlite3.connect(":memory:")
    repo = SQLiteWordsRepo(conn)
    app = App(routes)
    app.dependency_overrides[SupportsWordsRepo] = lambda: repo
    return app

def test_add_word_endpoint() -> None:
    # this demonstrates how easy it is to swap
    # out an implementation with this pattern
    words: List[str] = []

    class TestWordsRepo(SupportsWordsRepo):
        def add_word(self, word: str) -> None:

    add_word(TestWordsRepo(), "hello")

    assert words == ["hello"]

Notice that we didn't have to add a verbose Depends(...) to our endpoint function since we are wiring WordsRepo up in our composition root.

This pattern also lends iteself natually to depending on abstractions: because you aren't forced to specify how WordsRepo should be built, it can be an abstract interface (SupportsWordsRepo, using typing.Protocol or abc.ABC), leaving you with a clean and testable endpoint handler that has no mention of the concrete implementation of SupportsWordsRepo that will be used at runtime

Running an ASIG server programmatically

If you are running your ASGI server programmatically you have control of the event loop, allowing you to intialize arbitrarily complex dependencies (for example, a database connection that requires an async context manager).

This also has the side effect of making ASGI lifespans in redundant since you can do anything a lifespan can yourself before starting the ASGI server.

Here is an example of this pattern using Uvicorn

import asyncpg  # type: ignore[import]
import uvicorn  # type: ignore[import]

from xpresso import App

async def main() -> None:
    app = App()
    async with asyncpg.create_pool(...) as pool:  # type: ignore
        app.dependency_overrides[asyncpg.Pool] = lambda: pool
        server = uvicorn.Server(uvicorn.Config(app))
        await server.serve()

There are many variations to this pattern, you should try different arrangements to find one that best fits your use case. For example, you could:

  • Splitting out your aggregate root into a build_app() function
  • Mix this manual wiring in the aggregate root with use of Depends()
  • Use Uvicorn's --factory CLI parameter or, factory=True) if you'd like to wire up your dependencies in a composition root but don't need to take control of the event loop or need to run under Gunicorn.