Dependents and the DependentBase

Most of these docs use Dependent 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 DependentBase, which you can find in di.api.dependencies. Dependent is just one possible implementation of the DependentBase.

You can easily build your own version of Dependent by inheriting from Dependent or DependentBase.

Here is an example that extracts headers from requests:

import inspect
from typing import Mapping, Optional, TypeVar

from di import Container, bind_by_type
from di.dependent import Dependent, 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) -> Dependent[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 Dependent(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(Dependent(lambda: valid_request, scope="request"), Request)
    ):
        solved = container.solve(
            Dependent(controller, scope="request"), scopes=["request"]
        )
    with container.enter_scope("request") as state:
        await solved.execute_async(executor=AsyncExecutor(), state=state)  # success

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

    with container.enter_scope("request") as state:
        try:
            await solved.execute_async(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 DependentBase is the implementation of JointDependent, which lets you schedule and execute dependencies together even if they are not directly connected by wiring:

from di import Container
from di.dependent import Dependent, JoinedDependent
from di.executors import SyncExecutor


class A:
    ...


class B:
    executed = False

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


def main():
    container = Container()
    dependent = JoinedDependent(
        Dependent(A, scope="request"),
        siblings=[Dependent(B, scope="request")],
    )
    solved = container.solve(dependent, scopes=["request"])
    with container.enter_scope("request") as state:
        a = solved.execute_sync(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 JoinedDependent leverages the DependentBase interface to tell di that B is a dependency of A even if B is not a parameter or otherwise related to A.