API documentation

Connectable

You can inherit from this class to make the dependency visible for DI without adding these empty methods.

Source code in src/magic_di/_connectable.py

class Connectable:
    """
    You can inherit from this class to make the dependency visible for DI
    without adding these empty methods.
    """

    async def __connect__(self) -> None: ...

    async def __disconnect__(self) -> None: ...

ConnectableProtocol

Bases: Protocol

Interface for injectable clients. Adding these methods to your class will allow it to be dependency injectable. The dependency injector uses duck typing to check that the class implements the interface. This means that you do not need to inherit from this protocol.

Source code in src/magic_di/_connectable.py

@runtime_checkable
class ConnectableProtocol(Protocol):
    """
    Interface for injectable clients.
    Adding these methods to your class will allow it to be dependency injectable.
    The dependency injector uses duck typing to check that the class
    implements the interface.
    This means that you do not need to inherit from this protocol.
    """

    async def __connect__(self) -> None: ...

    async def __disconnect__(self) -> None: ...

DependencyInjector

Source code in src/magic_di/_injector.py

class DependencyInjector:
    def __init__(
        self,
        bindings: dict[type, type] | None = None,
        logger: logging.Logger = logger,
    ):
        self.bindings = bindings or {}
        self.logger: logging.Logger = logger

        self._deps = SingletonDependencyContainer()
        self._postponed: list[Callable[..., T]] = []
        self._lock = Lock()

    def inject(self, obj: Callable[..., T]) -> Callable[..., T]:
        """
        Inject dependencies into a class/function.
        This method is idempotent, always returns the same instance

        Args:
            obj (Callable[..., T]): The class/function to inject dependencies into.

        Returns:
            Callable[..., T]: Partial for class/function with dependencies injected.
        """
        obj = self._unwrap_type_hint(obj)  # type: ignore[arg-type]

        if dep := self._deps.get(obj):
            return dep

        signature = self.inspect(obj)

        clients: dict[str, object] = {}
        for name, dep in signature.deps.items():
            clients[name] = self.inject(dep)()

        if signature.injector_arg is not None:
            clients[signature.injector_arg] = self

        try:
            return self._deps.add(obj, **clients)
        except TypeError as exc:
            raise InjectionError(obj, signature) from exc

    def inspect(self, obj: AnyObject) -> Signature[AnyObject]:
        try:
            hints: dict[str, type[Any]] = get_type_hints(obj)
            hints_with_extras = get_type_hints(obj, include_extras=True)

            if not hints:
                return Signature(obj, is_injectable=bool(is_connectable(obj)))

            if inspect.ismethod(obj):
                hints.pop("self", None)

            hints.pop("return", None)

            signature = Signature(obj, is_injectable=bool(is_connectable(obj)))

            for name, hint_ in hints.items():
                hint = self._unwrap_type_hint(hint_)
                hint_with_extra = hints_with_extras[name]

                if is_injector(hint):
                    signature.injector_arg = name
                elif not is_connectable(hint) and not is_forcefully_marked_as_injectable(
                    hint_with_extra,
                ):
                    signature.kwargs[name] = hint
                else:
                    signature.deps[name] = hint

        except Exception as exc:
            raise InspectionError(obj) from exc

        return signature

    async def connect(self) -> None:
        """
        Connect all injected dependencies
        """
        # unpack to create copy of list
        for postponed in [*self._postponed]:
            # First, we need to create instances of postponed injection
            # in order to connect them.
            self.inject(postponed)

        for cls, instance in self._deps.iter_instances():
            if connectable_instance := is_connectable(instance):
                self.logger.debug("Connecting %s...", cls.__name__)
                await connectable_instance.__connect__()

    async def disconnect(self) -> None:
        """
        Disconnect all injected dependencies
        """
        for cls, instance in self._deps.iter_instances(reverse=True):
            if connectable_instance := is_connectable(instance):
                try:
                    await connectable_instance.__disconnect__()
                except Exception:
                    self.logger.exception("Failed to disconnect %s", cls.__name__)

    def get_dependencies_by_interface(
        self,
        interface: Callable[..., AnyObject],
    ) -> Iterable[AnyObject]:
        """
        Get all injected dependencies that implement a particular interface.
        """
        for _, instance in self._deps.iter_instances():
            if safe_is_instance(instance, interface):  # type: ignore[arg-type]
                yield instance  # type: ignore[misc]

    async def __aenter__(self) -> DependencyInjector:  # noqa: PYI034
        await self.connect()
        return self

    async def __aexit__(self, *args: object, **kwargs: Any) -> None:
        await self.disconnect()

    def iter_deps(self) -> Iterable[object]:
        instance: object

        for _, instance in self._deps.iter_instances():
            yield instance

    def lazy_inject(self, obj: Callable[..., T]) -> Callable[..., T]:
        """
        Lazily inject dependencies into a class or function.

        Args:
            obj (Callable[..., T]): The class or function to inject dependencies into.

        Returns:
            Callable[..., T]: A function that, when called,
            will inject the dependencies and
            return the injected class instance or function.
        """

        with self._lock:
            self._postponed.append(obj)  # type: ignore[arg-type]
            # incompatible type "Callable[..., T]"; expected "Callable[..., T]"

        injected: T | None = None

        def inject() -> T:
            nonlocal injected

            if injected is not None:
                return injected

            injected = self.inject(obj)()
            return injected

        return cast(type[T], inject)

    def bind(self, bindings: dict[type, type]) -> None:
        """
        Bind new bindings to the injector.

        This method is used to add new bindings to the injector.
        Bindings are a dictionary where the keys are the
        classes used in dependencies type hints,
        and the values are the classes that should replace them.

        For example, if you have a class `Foo` that depends on an interface `Bar`,
        and you have a class `BarImpl` that implements `Bar`,
        you would add a binding like this: `injector.bind({Bar: BarImpl})`.
        Then, whenever `Foo` is injected,
        it will receive an instance of `BarImpl` instead of `Bar`.

        If a binding for a particular class or type already exists,
        this method will update that binding with the new value

        Args:
            bindings (dict[type, type]): The bindings to add.
            This should be a dictionary where the keys are
            classes and the values are the classes that should replace them.
        """
        with self._lock:
            self.bindings = self.bindings | bindings

    @contextmanager
    def override(self, bindings: dict[type, type]) -> Iterator[None]:
        """
        Temporarily override the bindings and dependencies of the injector.

        Args:
            bindings (dict): The bindings to use for the override.
        """
        with self._lock:
            actual_deps = self._deps
            actual_bindings = self.bindings

            self._deps = SingletonDependencyContainer()
            self.bindings = self.bindings | bindings

        try:
            yield
        finally:
            with self._lock:
                self._deps = actual_deps
                self.bindings = actual_bindings

    def _unwrap_type_hint(self, obj: type[AnyObject]) -> type[AnyObject]:
        obj = get_cls_from_optional(obj)
        return self.bindings.get(obj, obj)

    def __hash__(self) -> int:
        """Injector is always unique"""
        return id(self)

__hash__() -> int

Injector is always unique

Source code in src/magic_di/_injector.py

def __hash__(self) -> int:
    """Injector is always unique"""
    return id(self)

bind(bindings: dict[type, type]) -> None

Bind new bindings to the injector.

This method is used to add new bindings to the injector. Bindings are a dictionary where the keys are the classes used in dependencies type hints, and the values are the classes that should replace them.

For example, if you have a class Foo that depends on an interface Bar, and you have a class BarImpl that implements Bar, you would add a binding like this: injector.bind({Bar: BarImpl}). Then, whenever Foo is injected, it will receive an instance of BarImpl instead of Bar.

If a binding for a particular class or type already exists, this method will update that binding with the new value

Parameters:

Name	Type	Description	Default
bindings	dict[type, type]	The bindings to add.	required

Source code in src/magic_di/_injector.py

def bind(self, bindings: dict[type, type]) -> None:
    """
    Bind new bindings to the injector.

    This method is used to add new bindings to the injector.
    Bindings are a dictionary where the keys are the
    classes used in dependencies type hints,
    and the values are the classes that should replace them.

    For example, if you have a class `Foo` that depends on an interface `Bar`,
    and you have a class `BarImpl` that implements `Bar`,
    you would add a binding like this: `injector.bind({Bar: BarImpl})`.
    Then, whenever `Foo` is injected,
    it will receive an instance of `BarImpl` instead of `Bar`.

    If a binding for a particular class or type already exists,
    this method will update that binding with the new value

    Args:
        bindings (dict[type, type]): The bindings to add.
        This should be a dictionary where the keys are
        classes and the values are the classes that should replace them.
    """
    with self._lock:
        self.bindings = self.bindings | bindings

connect() -> None async

Connect all injected dependencies

Source code in src/magic_di/_injector.py

async def connect(self) -> None:
    """
    Connect all injected dependencies
    """
    # unpack to create copy of list
    for postponed in [*self._postponed]:
        # First, we need to create instances of postponed injection
        # in order to connect them.
        self.inject(postponed)

    for cls, instance in self._deps.iter_instances():
        if connectable_instance := is_connectable(instance):
            self.logger.debug("Connecting %s...", cls.__name__)
            await connectable_instance.__connect__()

disconnect() -> None async

Disconnect all injected dependencies

Source code in src/magic_di/_injector.py

async def disconnect(self) -> None:
    """
    Disconnect all injected dependencies
    """
    for cls, instance in self._deps.iter_instances(reverse=True):
        if connectable_instance := is_connectable(instance):
            try:
                await connectable_instance.__disconnect__()
            except Exception:
                self.logger.exception("Failed to disconnect %s", cls.__name__)

get_dependencies_by_interface(interface: Callable[..., AnyObject]) -> Iterable[AnyObject]

Get all injected dependencies that implement a particular interface.

Source code in src/magic_di/_injector.py

def get_dependencies_by_interface(
    self,
    interface: Callable[..., AnyObject],
) -> Iterable[AnyObject]:
    """
    Get all injected dependencies that implement a particular interface.
    """
    for _, instance in self._deps.iter_instances():
        if safe_is_instance(instance, interface):  # type: ignore[arg-type]
            yield instance  # type: ignore[misc]

inject(obj: Callable[..., T]) -> Callable[..., T]

Inject dependencies into a class/function. This method is idempotent, always returns the same instance

Parameters:

Name	Type	Description	Default
obj	Callable[..., T]	The class/function to inject dependencies into.	required

Returns:

Type	Description
Callable[..., T]	Callable[..., T]: Partial for class/function with dependencies injected.

Source code in src/magic_di/_injector.py

def inject(self, obj: Callable[..., T]) -> Callable[..., T]:
    """
    Inject dependencies into a class/function.
    This method is idempotent, always returns the same instance

    Args:
        obj (Callable[..., T]): The class/function to inject dependencies into.

    Returns:
        Callable[..., T]: Partial for class/function with dependencies injected.
    """
    obj = self._unwrap_type_hint(obj)  # type: ignore[arg-type]

    if dep := self._deps.get(obj):
        return dep

    signature = self.inspect(obj)

    clients: dict[str, object] = {}
    for name, dep in signature.deps.items():
        clients[name] = self.inject(dep)()

    if signature.injector_arg is not None:
        clients[signature.injector_arg] = self

    try:
        return self._deps.add(obj, **clients)
    except TypeError as exc:
        raise InjectionError(obj, signature) from exc

lazy_inject(obj: Callable[..., T]) -> Callable[..., T]

Lazily inject dependencies into a class or function.

Parameters:

Name	Type	Description	Default
obj	Callable[..., T]	The class or function to inject dependencies into.	required

Returns:

Type	Description
Callable[..., T]	Callable[..., T]: A function that, when called,
Callable[..., T]	will inject the dependencies and
Callable[..., T]	return the injected class instance or function.

Source code in src/magic_di/_injector.py

def lazy_inject(self, obj: Callable[..., T]) -> Callable[..., T]:
    """
    Lazily inject dependencies into a class or function.

    Args:
        obj (Callable[..., T]): The class or function to inject dependencies into.

    Returns:
        Callable[..., T]: A function that, when called,
        will inject the dependencies and
        return the injected class instance or function.
    """

    with self._lock:
        self._postponed.append(obj)  # type: ignore[arg-type]
        # incompatible type "Callable[..., T]"; expected "Callable[..., T]"

    injected: T | None = None

    def inject() -> T:
        nonlocal injected

        if injected is not None:
            return injected

        injected = self.inject(obj)()
        return injected

    return cast(type[T], inject)

override(bindings: dict[type, type]) -> Iterator[None]

Temporarily override the bindings and dependencies of the injector.

Parameters:

Name	Type	Description	Default
bindings	dict	The bindings to use for the override.	required

Source code in src/magic_di/_injector.py

@contextmanager
def override(self, bindings: dict[type, type]) -> Iterator[None]:
    """
    Temporarily override the bindings and dependencies of the injector.

    Args:
        bindings (dict): The bindings to use for the override.
    """
    with self._lock:
        actual_deps = self._deps
        actual_bindings = self.bindings

        self._deps = SingletonDependencyContainer()
        self.bindings = self.bindings | bindings

    try:
        yield
    finally:
        with self._lock:
            self._deps = actual_deps
            self.bindings = actual_bindings

celery

fastapi

Package containing tools for integrating Dependency Injector with FastAPI framework.

inject_app(app: FastAPI, *, injector: DependencyInjector | None = None, use_deprecated_events: bool = False) -> FastAPI

Inject dependencies into a FastAPI application using the provided injector.

This function sets up the FastAPI application to connect and disconnect the injector at startup and shutdown, respectively.

This ensures that all dependencies are properly connected when the application starts, and properly disconnected when the application shuts down.

If no injector is provided, a default injector is used.

Parameters:

Name	Type	Description	Default
app	FastAPI	The FastAPI application	required
injector	DependencyInjector	The injector to use for dependency injection. If not provided, a default injector will be used.	None
use_deprecated_events	bool	Indicate whether the app should be injected with starlette events (which are deprecated) or use app lifespans. Important: Use this flag only if you still use app events, because if lifespans are defined, events will be ignored by Starlette.	False

Returns:

Name	Type	Description
FastAPI	FastAPI	The FastAPI application with dependencies injected.

Source code in src/magic_di/fastapi/_app.py

def inject_app(
    app: FastAPI,
    *,
    injector: DependencyInjector | None = None,
    use_deprecated_events: bool = False,
) -> FastAPI:
    """
    Inject dependencies into a FastAPI application using the provided injector.

    This function sets up the FastAPI application to connect and disconnect the injector
    at startup and shutdown, respectively.

    This ensures that all dependencies are properly
    connected when the application starts, and properly disconnected when the application
    shuts down.

    If no injector is provided, a default injector is used.

    Args:
        app (FastAPI): The FastAPI application

        injector (DependencyInjector, optional): The injector to use for dependency injection.
                                                 If not provided, a default injector will be used.

        use_deprecated_events (bool, optional): Indicate whether the app should be injected
                                                with starlette events (which are deprecated)
                                                or use app lifespans.
                                                Important: Use this flag only
                                                if you still use app events,
                                                because if lifespans are defined,
                                                events will be ignored by Starlette.

    Returns:
        FastAPI: The FastAPI application with dependencies injected.
    """
    injector = injector or DependencyInjector()

    def collect_deps() -> None:
        _collect_dependencies(injector, app.router)

    app.state.dependency_injector = injector

    if use_deprecated_events:
        _inject_app_with_events(app, collect_deps_fn=collect_deps)
    else:
        _inject_app_with_lifespan(app, collect_deps_fn=collect_deps)

    return app

healthcheck

DependenciesHealthcheck dataclass

Bases: Connectable

Injectable Healthcheck component that pings all injected dependencies that implement the PingableProtocol

Example usage:

from app.components.services.health import DependenciesHealthcheck

async def main(redis: Redis, deps_healthcheck: DependenciesHealthcheck) -> None:
    await deps_healthcheck.ping_dependencies()  # redis will be pinged if it has method __ping__

inject_and_run(main)

Source code in src/magic_di/healthcheck.py

@dataclass
class DependenciesHealthcheck(Connectable):
    """
    Injectable Healthcheck component that pings all injected dependencies
    that implement the PingableProtocol

    Example usage:

    ``` py
    from app.components.services.health import DependenciesHealthcheck

    async def main(redis: Redis, deps_healthcheck: DependenciesHealthcheck) -> None:
        await deps_healthcheck.ping_dependencies()  # redis will be pinged if it has method __ping__

    inject_and_run(main)
    ```
    """

    injector: DependencyInjector

    async def ping_dependencies(self, max_concurrency: int = 1) -> None:
        """
        Ping all dependencies that implement the PingableProtocol

        :param max_concurrency: Maximum number of concurrent pings
        """
        tasks: set[Future[Any]] = set()

        try:
            for dependency in self.injector.get_dependencies_by_interface(PingableProtocol):
                future = asyncio.ensure_future(self.ping(dependency))
                tasks.add(future)

                if len(tasks) >= max_concurrency:
                    tasks, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

            if tasks:
                await asyncio.gather(*tasks)
                tasks = set()

        finally:
            for task in tasks:
                task.cancel()

            if tasks:
                with suppress(asyncio.CancelledError):
                    await asyncio.gather(*tasks)

    async def ping(self, dependency: PingableProtocol) -> None:
        """
        Ping a single dependency

        :param dependency: Dependency to ping
        """
        dependency_name = dependency.__class__.__name__
        self.injector.logger.debug("Pinging dependency %s...", dependency_name)

        await dependency.__ping__()

        self.injector.logger.debug("Dependency %s is healthy", dependency_name)

ping(dependency: PingableProtocol) -> None async

Ping a single dependency

:param dependency: Dependency to ping

Source code in src/magic_di/healthcheck.py

async def ping(self, dependency: PingableProtocol) -> None:
    """
    Ping a single dependency

    :param dependency: Dependency to ping
    """
    dependency_name = dependency.__class__.__name__
    self.injector.logger.debug("Pinging dependency %s...", dependency_name)

    await dependency.__ping__()

    self.injector.logger.debug("Dependency %s is healthy", dependency_name)

ping_dependencies(max_concurrency: int = 1) -> None async

Ping all dependencies that implement the PingableProtocol

:param max_concurrency: Maximum number of concurrent pings

Source code in src/magic_di/healthcheck.py

async def ping_dependencies(self, max_concurrency: int = 1) -> None:
    """
    Ping all dependencies that implement the PingableProtocol

    :param max_concurrency: Maximum number of concurrent pings
    """
    tasks: set[Future[Any]] = set()

    try:
        for dependency in self.injector.get_dependencies_by_interface(PingableProtocol):
            future = asyncio.ensure_future(self.ping(dependency))
            tasks.add(future)

            if len(tasks) >= max_concurrency:
                tasks, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

        if tasks:
            await asyncio.gather(*tasks)
            tasks = set()

    finally:
        for task in tasks:
            task.cancel()

        if tasks:
            with suppress(asyncio.CancelledError):
                await asyncio.gather(*tasks)

testing

InjectableMock

Bases: AsyncMock

You can use this mock to override dependencies in tests and use AsyncMock instead of a real class instance

Example:

@pytest.fixture()
def client():
  injector = DependencyInjector()

  with injector.override({Service: InjectableMock().mock_cls}):
    with TestClient(app) as client:
        yield client

def test_http_handler(client):
  resp = client.post('/hello-world')

  assert resp.status_code == 200

Source code in src/magic_di/testing.py

class InjectableMock(AsyncMock):
    """
    You can use this mock to override dependencies in tests
    and use AsyncMock instead of a real class instance

    Example:
    ``` py
    @pytest.fixture()
    def client():
      injector = DependencyInjector()

      with injector.override({Service: InjectableMock().mock_cls}):
        with TestClient(app) as client:
            yield client

    def test_http_handler(client):
      resp = client.post('/hello-world')

      assert resp.status_code == 200
    ```
    """

    @property
    def mock_cls(self) -> type[ConnectableProtocol]:
        return get_injectable_mock_cls(self)

    async def __connect__(self) -> None: ...

    async def __disconnect__(self) -> None: ...

    def __call__(self, *args: Any, **kwargs: Any) -> InjectableMock:
        return self.__class__(*args, **kwargs)

utils

inject_and_run(fn: Callable[..., T], injector: DependencyInjector | None = None) -> T

This function takes a callable, injects dependencies into it using the provided injector, and then runs the function. If the function is a coroutine, it will be awaited.

The function itself is not asynchronous, but it uses asyncio.run to run the internal coroutine, so it is suitable for use in synchronous code.

Parameters:

Name	Type	Description	Default
fn	Callable	The function into which dependencies will be injected. This can be a regular function or a coroutine function.	required
injector	DependencyInjector	The injector to use for dependency injection. If not provided, a default injector will be used.	None

Returns:

Name	Type	Description
Any	T	The return value of the function fn after dependency injection and execution.

Source code in src/magic_di/utils.py

def inject_and_run(
    fn: Callable[..., T],
    injector: DependencyInjector | None = None,
) -> T:
    """
    This function takes a callable, injects dependencies into it using the provided injector,
    and then runs the function. If the function is a coroutine, it will be awaited.

    The function itself is not asynchronous, but it uses asyncio.run
    to run the internal coroutine, so it is suitable for use in synchronous code.

    Args:
        fn (Callable): The function into which dependencies will be injected. This can be
                       a regular function or a coroutine function.
        injector (DependencyInjector, optional): The injector to use for dependency injection.
                                                 If not provided, a default injector will be used.

    Returns:
        Any: The return value of the function `fn` after dependency injection and execution.

    Raises:
        Any exceptions raised by the function `fn` or the injector will be propagated up to
        the caller of this function.
    """
    injector = injector or DependencyInjector()

    async def run() -> T:
        injected = injector.inject(fn)

        async with injector:
            if inspect.iscoroutinefunction(fn):
                return await injected()  # type: ignore[misc,no-any-return]

            return injected()

    return asyncio.run(run())