Dependants and the DependantBase

Most of these docs use Dependant as the main marker for dependencies. But the container doesn't actually know about either of these two things! In fact, the container only knows about the DependantBase, which you can find in di.api.dependencies. Dependant is just one possible implementation of the DependantBase.

You can easily build your own version of Dependant by inheriting from Dependant or DependantBase.

Here is an example that extracts headers from requests:

import inspect
from typing import Mapping, Optional, TypeVar

from di.container import Container, bind_by_type
from di.dependant import Dependant, Marker
from di.executors import AsyncExecutor
from di.typing import Annotated


class Request:
    def __init__(self, headers: Mapping[str, str]) -> None:
        self.headers = {k.lower(): v for k, v in headers.items()}


class Header(Marker):
    def __init__(self, alias: Optional[str]) -> None:
        self.alias = alias
        super().__init__(call=None, scope="request", use_cache=False)

    def register_parameter(self, param: inspect.Parameter) -> Dependant[str]:
        if self.alias is not None:
            name = self.alias
        else:
            name = param.name.replace("_", "-")

        def get_header(request: Annotated[Request, Marker()]) -> str:
            return param.annotation(request.headers[name])

        return Dependant(get_header, scope="request")


T = TypeVar("T")

FromHeader = Annotated[T, Header(alias=None)]


async def web_framework() -> None:
    container = Container()

    valid_request = Request(headers={"x-header-one": "one", "x-header-two": "2"})
    with container.bind(
        bind_by_type(Dependant(lambda: valid_request, scope="request"), Request)
    ):
        solved = container.solve(
            Dependant(controller, scope="request"), scopes=["request"]
        )
    with container.enter_scope("request") as state:
        await container.execute_async(
            solved, executor=AsyncExecutor(), state=state
        )  # success

    invalid_request = Request(headers={"x-header-one": "one"})
    with container.bind(
        bind_by_type(Dependant(lambda: invalid_request, scope="request"), Request)
    ):
        solved = container.solve(
            Dependant(controller, scope="request"), scopes=["request"]
        )

    with container.enter_scope("request") as state:
        try:
            await container.execute_async(
                solved, executor=AsyncExecutor(), state=state
            )  # fails
        except KeyError:
            pass
        else:
            raise AssertionError(
                "This call should have failed because x-header-two is missing"
            )


def controller(
    x_header_one: FromHeader[str],
    header_two_val: Annotated[int, Header(alias="x-header-two")],
) -> None:
    """This is the only piece of user code"""
    assert x_header_one == "one"
    assert header_two_val == 2

Another good example of the flexibility provided by DependantBase is the implementation of JointDependant, which lets you schedule and execute dependencies together even if they are not directly connected by wiring:

from di.container import Container
from di.dependant import Dependant, JoinedDependant
from di.executors import SyncExecutor


class A:
    ...


class B:
    executed = False

    def __init__(self) -> None:
        B.executed = True


def main():
    container = Container()
    dependant = JoinedDependant(
        Dependant(A, scope="request"),
        siblings=[Dependant(B, scope="request")],
    )
    solved = container.solve(dependant, scopes=["request"])
    with container.enter_scope("request") as state:
        a = container.execute_sync(solved, executor=SyncExecutor(), state=state)
    assert isinstance(a, A)
    assert B.executed

Here B is executed even though A does not depend on it. This is because JoinedDependant leverages the DependantBase interface to tell di that B is a dependency of A even if B is not a parameter or otherwise related to A.