di: pythonic dependency injection
Tags: python, programming, testing, opensource
Published: 2021-10-01 16:12:42

In this article, I would like to introduce [di], a dependency injection framework that seeks to be:

- **Intuitive**: simple things are easy, complex things are possible
- **Succinct**: low boilerplate, can run your code directly via autowiring
- **Correct**: tested with MyPy; exports valid type annotations
- **Powerful**: with lifecycles, shared dependencies and a flexible API
- **Performant**: supporting async dependencies and concurrent execution

## Introduction to dependency injection

Dependency injection is a technique for organizing larger applications by handing over control of your dependencies to "something" (often a framework) that assembles them for you.
It's a complicated topic, so maybe it's better to just look at an example.
We'll build our example around a pretty common scenario: you have a class that needs to make web requests, so it needs an HTTP client (we'll use httpx).

```python
from httpx import Client


class WeatherClient:
    def __init__(self) -> None:
        self.client = Client()
    def get_temp(self, lat: float, long: float) -> float:
        self.client.get(...)
        ...
```

This class does _not_ use dependency injection.
The problem arises when we want to test this class: we probably don't want to make real web requests in our unit test.

Luckily, httpx offers the ability to use [mock transports]:

```python
import httpx

def handler(request: httpx.Request) -> httpx.Response:
    return httpx.Response(200, json={"temp": 43.21})

client = httpx.Client(transport=httpx.MockTransport(handler))
```

But we have a problem: how do we get our `client` using a mock transport into our `WeatherClient`?
One approach often used is [monkey patching].
While monkey patching has its use cases, it can also lead to a lot of very confusing problems and messy code, especially in a larger project.

Dependency injection gives us an alternative: have `WeatherClient` accept an *instance* of `httpx.Client`:

```python
from typing import Optional

from httpx import Client


class WeatherClient:
    def __init__(self, client: Optional[Client] = None) -> None:
        self.client = client or Client()
```

This form of dependency injection is called *constructor injection* because we inject the dependency in the constructor / `__init__` method.

In this case, we chose to add a `None` default so that we can detect when the caller did not include a `Client` and so we can make a default one.
This is largely a matter of preference, and depends on the project you are working on.
For a library, this is probably a good idea.
For an application, maybe not so much.

**Note**: did you notice that `httpx.Client` itself uses dependency injection for the `transport` parameter? Neat huh!

### The problems

As per above, dependency injection is not as crazy as it sounds, it can be simple.
And you're probably already using it in libraries like `httpx`.

*But what are the downsides?*

Well the main issue is that when you end up with deeply nested / a lot of dependencies, as would tend to happen in a larger project, you can end up with some really ugly constructor calls, or you may need the same dependency in multiple places, which means you need a temporary variable, etc.

Basically, the code to *assemble* and keep track of all of these dependencies can become a headache in and of itself.

Enter: dependency injection frameworks.

Dependency injection frameworks seek to solve this problem by abstracting away the boilerplate of creating and connecting dependencies together.

### Dependency injection frameworks

Although dependency injection frameworks come in many shapes and sizes, in principle they all provide similar functionality:

- Register dependencies and providers for those dependencies (i.e. tell the framework how to build your dependencies)
- Inject your dependencies for you

Some frameworks also have more advanced features such as:

- Autowiring: transitive dependencies are discovered from parameter names or type annotations
- Lifecycles: the ability to hook into creation, deletion and error events during the lifecycle of your dependency.
- Scopes: singleton dependencies or dependencies that are re-used within each call if they are needed in multiple places.

`di` is one of the latter frameworks, offering all of these features and more.

## Using `di`

Let's start with a simple example.
We want to build a web app that uses our `WeatherClient`.

```python
import os
from dataclasses import dataclass, field

from xyz import App


def url_from_env() -> str:
    return os.environ["WEATHER_URL"]


@dataclass
class AppConfig:
     weather_url: str = field(default_factory=url_from_env)


class MyApp:
    def __init__(self, weather_client: WeatherClient) -> None:
    self.weather_client = weather_client
```

Building this by hand might look something like:

```python
app = App(
    weather_client=WeatherClient(
        client=httpx.Client(
            ...
        )
    ),
    config=AppConfig(
        ...
    )
)
```

Using `di`:

```python
from di import Container, Dependant


container = Container()
app = container.execute_sync(
    container.solve(Dependant(App))
)
```

Notice that it doesn't matter how deep the dependencies go or how many there are, we only ever said "we want an App instance".

What about the injection of the `Client`?

```python
client = Client(
    transport=MockTransport(handler=handler)
)
container.bind(Dependant(lambda: client), Client)
app = container.execute_sync(
    container.solve(Dependant(App))
)
```

Again, the important thing here is that `Client` could be nested 10 layers of dependencies deep and the above code would not change at all.

### Advanced functionality

Aside from the basic usage above, `di` support a lot more functionality.

Here are some snippets demonstrating the functionality, but for full fledged examples and more explanation, see our [docs].

#### Dependency sharing

```python
from di import Container, Depends, Dependant

def func() -> object:
    return object()

def dependant(
    one: object = Depends(func),
    two: object = Depends(func),
    three: object = Depends(func, share=False)
) -> None:
    assert one is two and two is not three

container = Container()
container.execute_sync(container.solve(Dependant(dependant)))
```

In this example, `func` is called twice (by default in two separate threads), once for `one` and `two`, which share the same value, and once for `three`.

#### Scopes and lifecycles

```python
from typing import Generator

from di import Container, Depends, Dependant

def func() -> Generator[int, None, None]:
   print("func startup")
   yield 1
   print("func shutdown")


def dependant(v: int = Depends(func, scope=1234)) -> int:
    print("computing")
    return v + 1

container = Container()
with container.enter_local_scope(1234):
    print("enter scope")
    res = container.execute_sync(
       container.solve(Dependant(dependant))
    )
    print("exit scope")

assert res == 2
```

This example will print:

```
enter scope
func startup
computing
func shutdown
exit scope
```

And is equivalent to:

```python
from contextlib import contextmanager


with contextmanager(func)() as value:
    dependant(v=value)
```

Scopes in `di` are arbitrary: any hashable value will do.
As long as you are consistent in declaring scopes, `di` will not care what you use. If "you" are a framework calling into user code, this gives you the freedom to define and manage your own scopes. For a web framework, these might be "app" and "request". For a CLI, there might be a single "call" scope. For a TUI, there could be an "app" and "event scope"

#### Solved dependencies

Often, you have to run the same dependency multiple times but aren't dynamically changing the dependency graph every execution (you would be dynamically changing the graph if you were doing `container.bind` between each call).

In these scenarios, `di` lets you save the entire computed dependency graph into a `SavedDependency` and execute the same graph multiple times.
In fact, this is what we've been doing already every time we called `container.solve`.


```python
# solving locks in binds, but not cached values or scopes
solved = container.solve(Dependant(dependency))
res1 = container.execute_sync(solved)
res2 = container.execute_sync(solved)
```

#### Async execution

Simply swap out `execute_sync` for `execute_async` for async support:

```python
from di import Container, Dependant


async def dep() -> int:
    return 1

async def main() -> None:
    container = Container()
    solved = container.solve(Dependant(dep))
    res = await container.execute_async(solved)
    assert res == 1
```

Like sync execution, async execution supports arbitrary callables as well as generator functions that get used as context managers.

#### DependantProtocol

The `DependantProtocol` interface provides an abstract representation of a dependency specification.
The Container uses this abstract representation to build dependency graphs.

This means that you can easily customize a lot of the behavior in `di` by overriding this API.
If you want to use some of the built in functionality and just tweak it slightly, you can inherit from `di.Dependant`, but you don't have to.

For example, to accept a default value instead of a callable:

```python
from di import Dependant
from di.types.providers import DependencyType
from di.types.scopes import Scope

class DefaultDependant(Dependant[DependencyType]):
    def __init__(
        self,
        default: DependencyType,
        scope: Scope,
        shared: bool,
    ) -> None:
        super().__init__(
            call=lambda: default,
            scope=scope,
            share=share,
         )
```

Now if there is no bind / override on the dependency, it will execute just like if it were being called directly.

#### Executors

One of the design principles of `di` is to isolate the work of the container into distinct steps:
- Wiring, which can be controlled via the DependantProtocol
- Solving, which is owned by the container
- Execution, which is managed via the Executor API

To this end, `di` allows you to provide your own executor, which only needs to take a list of groups of tasks (callable functions that take no arguments) and execute them in the specified order.

The Executor API is as follows:

```python
import typing

ResultType = typing.TypeVar("ResultType")

Task = typing.Union[
    typing.Callable[[], None],
    typing.Callable[[], typing.Awaitable[None]]
]


class SyncExecutor(typing.Protocol):
    def execute_sync(
        self,
        tasks: typing.List[typing.List[Task]],
        get_result: typing.Callable[[], ResultType],
    ) -> ResultType:
        ...


class AsyncExecutor(typing.Protocol):
    async def execute_async(
        self,
        tasks: typing.List[typing.List[Task]],
        get_result: typing.Callable[[], ResultType],
    ) -> ResultType:
        ...
```

The default executor (`di.executors.DefaultExecutor`) supports both sync and async execution.
Dependencies are executed concurrently (via threads in sync execution and tasks in async execution) for dependencies that can be executed in concurrently.

A simple SyncExecutor implementation might look like:

```python
class SimpleExecutor(SyncExecutor):
    def execute_sync(
        self,
        tasks: typing.List[typing.List[Task]],
        get_result: typing.Callable[[], ResultType],
    ) -> ResultType:
        for group in tasks:
            for task in group:
                res = task()
                if inspect.isawaitable(res):
                    raise TypeError
         return get_result()
```

Note that `SyncExecutor` may be handed async functions as tasks.
The sync/async portion refers to the API of the executor itself, not the the tasks it may be handed.
It is up to the implementation to raise an exception or execute the async task somehow.

## Integration into other libraries

While `di` should be simple enough that it can be used by end users, it shines when integrated into a framework or library that fully embraces inversion of control.

To this end, we provide [sample integrations], currently only for Starlette and Textual, but more ideas are welcome!

## Performance

We provide some [basic benchmarks comparing to FastAPI].

The results strongly depend on the size of the DAG.

For a relatively large but not completely absurd DAG of 12 vertices with each vertex having 0-3 edges, `di` performs slightly faster than FastAPI (a couple percent per request).

For larger DAGs, especially where `di` is able to execute dependencies concurrently, `di` can be 10x faster than FastAPI.
To be fair, this is not a real world scenario, and FastAPI does a lot more than the Starlette `di` integration is doing (keeping track of stuff for OpenAPI, etc.).
If these benchmarks are unfair for other reasons, please let me know.

## Conclusion

Dependency injection and inversion of control are powerful techniques, but it can get complicated.
Python has shown time and time again that it can take complicated things and make them simple for users.
FastAPI is one of the best examples of this.

I hope that `di` can help other libraries, frameworks or projects use inversion of control and dependency injection without having to go through the pain of writing their own dependency injection system.

[di]: https://www.adriangb.com/di/
[httpx]: https://www.python-httpx.org
[mock transports]: https://www.python-httpx.org/advanced/#mock-transports
[monkey patching]: https://docs.python.org/3/library/unittest.mock.html#patch
[docs]: https://www.adriangb.com/di/
[sample integrations]: https://www.adriangb.com/di/latest/integrations/
[basic benchmarks comparing to FastAPI]: https://github.com/adriangb/di/tree/main/benchmarks