Examples

Simple Example

Here is a simple example of how di works:

from dataclasses import dataclass

from di.container import Container
from di.dependent import Dependent
from di.executors import SyncExecutor


class A:
    ...


class B:
    ...


@dataclass
class C:
    a: A
    b: B


def main():
    container = Container()
    solved = container.solve(Dependent(C, scope="request"), scopes=["request"])
    with container.enter_scope("request") as state:
        c = container.execute_sync(solved, executor=SyncExecutor(), state=state)
    assert isinstance(c, C)
    assert isinstance(c.a, A)
    assert isinstance(c.b, B)

You will notice that di "auto-wired" C: we didn't have to tell it that C depends on A and B, or how to construct A and B, it was all inferred from type annotations.

In the wiring and provider registration chapters, you'll see how you can customize this behavior to tell di how to inject things like abstract interfaces or function return values.

In-depth example

With this background in place, let's dive into a more in-depth example.

In this example, we'll look at what it would take for a web framework to provide dependency injection to its users via di.

Let's start by looking at the User's code.

from typing import Any, Callable

from di.container import Container
from di.dependent import Dependent
from di.executors import SyncExecutor


# Framework code
class Request:
    def __init__(self, value: int) -> None:
        self.value = value


class App:
    def __init__(self, controller: Callable[..., Any]) -> None:
        self.container = Container()
        self.solved = self.container.solve(
            Dependent(controller, scope="request"),
            scopes=["request"],
        )
        self.executor = SyncExecutor()

    def run(self, request: Request) -> int:
        with self.container.enter_scope("request") as state:
            return self.container.execute_sync(
                self.solved,
                values={Request: request},
                executor=self.executor,
                state=state,
            )


# User code
class MyClass:
    def __init__(self, request: Request) -> None:
        self.value = request.value

    def add(self, value: int) -> int:
        return self.value + value


def controller(myobj: MyClass) -> int:
    return myobj.add(1)


def main() -> None:
    app = App(controller)
    resp = app.run(Request(1))
    assert resp == 2
    resp = app.run(Request(2))
    assert resp == 3


if __name__ == "__main__":
    main()

As a user, you have very little boilerplate. In fact, there is not a single line of code here that is not transmitting information.

Now let's look at the web framework side of things. This part can get a bit complex, but it's okay because it's written once, in a library.

First, we'll need to create a Container instance. This would be tied to the App or Router instance of the web framework.

from typing import Any, Callable

from di.container import Container
from di.dependent import Dependent
from di.executors import SyncExecutor


# Framework code
class Request:
    def __init__(self, value: int) -> None:
        self.value = value


class App:
    def __init__(self, controller: Callable[..., Any]) -> None:
        self.container = Container()
        self.solved = self.container.solve(
            Dependent(controller, scope="request"),
            scopes=["request"],
        )
        self.executor = SyncExecutor()

    def run(self, request: Request) -> int:
        with self.container.enter_scope("request") as state:
            return self.container.execute_sync(
                self.solved,
                values={Request: request},
                executor=self.executor,
                state=state,
            )


# User code
class MyClass:
    def __init__(self, request: Request) -> None:
        self.value = request.value

    def add(self, value: int) -> int:
        return self.value + value


def controller(myobj: MyClass) -> int:
    return myobj.add(1)


def main() -> None:
    app = App(controller)
    resp = app.run(Request(1))
    assert resp == 2
    resp = app.run(Request(2))
    assert resp == 3


if __name__ == "__main__":
    main()

Next we solve all of our endpoints/controllers (in this case just a single one). This should happen once, maybe at application startup, and then you should save the solved object, which contains all the information necessary to execute the dependency (dependency being in this case the user's endpoint/controller function). This is very important for performance: we want to do the least amount of work possible for each incoming request.

from typing import Any, Callable

from di.container import Container
from di.dependent import Dependent
from di.executors import SyncExecutor


# Framework code
class Request:
    def __init__(self, value: int) -> None:
        self.value = value


class App:
    def __init__(self, controller: Callable[..., Any]) -> None:
        self.container = Container()
        self.solved = self.container.solve(
            Dependent(controller, scope="request"),
            scopes=["request"],
        )
        self.executor = SyncExecutor()

    def run(self, request: Request) -> int:
        with self.container.enter_scope("request") as state:
            return self.container.execute_sync(
                self.solved,
                values={Request: request},
                executor=self.executor,
                state=state,
            )


# User code
class MyClass:
    def __init__(self, request: Request) -> None:
        self.value = request.value

    def add(self, value: int) -> int:
        return self.value + value


def controller(myobj: MyClass) -> int:
    return myobj.add(1)


def main() -> None:
    app = App(controller)
    resp = app.run(Request(1))
    assert resp == 2
    resp = app.run(Request(2))
    assert resp == 3


if __name__ == "__main__":
    main()

Finally, we execute the endpoint for each incoming request:

from typing import Any, Callable

from di.container import Container
from di.dependent import Dependent
from di.executors import SyncExecutor


# Framework code
class Request:
    def __init__(self, value: int) -> None:
        self.value = value


class App:
    def __init__(self, controller: Callable[..., Any]) -> None:
        self.container = Container()
        self.solved = self.container.solve(
            Dependent(controller, scope="request"),
            scopes=["request"],
        )
        self.executor = SyncExecutor()

    def run(self, request: Request) -> int:
        with self.container.enter_scope("request") as state:
            return self.container.execute_sync(
                self.solved,
                values={Request: request},
                executor=self.executor,
                state=state,
            )


# User code
class MyClass:
    def __init__(self, request: Request) -> None:
        self.value = request.value

    def add(self, value: int) -> int:
        return self.value + value


def controller(myobj: MyClass) -> int:
    return myobj.add(1)


def main() -> None:
    app = App(controller)
    resp = app.run(Request(1))
    assert resp == 2
    resp = app.run(Request(2))
    assert resp == 3


if __name__ == "__main__":
    main()

When we do this, we provide the Request instance as a value. This means that di does not introspect at all into the Request to figure out how to build it, it just hands the value off to anything that requests it. You can also directly register providers, which is covered in the provider registration section of the docs.

You'll also notice the executor parameter. As you'll see in the [architecture] chapter, one of the fundamental design principles in di is to decouple wiring, solving and execution. This makes it trivial to, for example, enable concurrent execution of dependencies using threads, asynchronous task groups or any other execution paradigm you want.