modern-di-taskiq 4.2.2

modern-di integration for taskiq.

Full guide: taskiq integration docs

Usage example: examples/

Installation

uv add modern-di-taskiq      # or: pip install modern-di-taskiq

Usage

setup_di stores the container on broker.state and opens it on WORKER_STARTUP. At shutdown it closes the container after every WORKER_SHUTDOWN handler and every middleware shutdown(), so your shutdown code can still resolve Scope.APP dependencies and their finalizers run last. FromDI resolves a provider (or type) into a task parameter through a taskiq dependency, which builds a Scope.REQUEST child container for the task and closes it when the task finishes. No per-task decorator is needed.

import typing

from modern_di import Container, Group, Scope, providers
from modern_di_taskiq import FromDI, setup_di
from taskiq import InMemoryBroker


class Settings:
    def __init__(self) -> None:
        self.greeting = "hello"


class Greeter:
    def __init__(self, settings: Settings) -> None:  # auto-injected by type
        self._settings = settings

    def greet(self, name: str) -> str:
        return f"{self._settings.greeting}, {name}"


class AppGroup(Group):
    settings = providers.Factory(Settings, scope=Scope.APP, cache=True)
    greeter = providers.Factory(Greeter, scope=Scope.REQUEST)


broker = InMemoryBroker()
container = Container(groups=[AppGroup])
setup_di(broker, container)
container.validate()  # optional fail-fast; must come after setup_di registers its providers


@broker.task
async def greet(
    name: str,
    greeter: typing.Annotated[Greeter, FromDI(Greeter)],  # resolve by type
) -> str:
    return greeter.greet(name)

The WORKER_STARTUP/WORKER_SHUTDOWN events fire when the broker's worker process starts and stops, so a script that calls tasks directly (like InMemoryBroker in a test) must drive the container lifecycle itself, for example with async with broker: .... Providers resolve synchronously, but the per-task Scope.REQUEST child is closed with close_async(), so REQUEST-scoped providers can have async finalizers. taskiq.TaskiqMessage is resolvable within DI via the pre-built taskiq_message_provider context provider.

Sharing a container

When one container serves several apps, for example a taskiq worker and a web app in the same process, only one of them should open and close it. Pass manage_lifespan=False and own the root container yourself:

container = Container(groups=[AppGroup])
setup_di(broker, container, manage_lifespan=False)


async def main() -> None:
    async with container:
        await broker.startup()
        ...
        await broker.shutdown()

The worker events then leave the root container alone. Each task still gets its own Scope.REQUEST child, closed when the task ends. APP-scoped finalizers run once, when the async with block exits.

API

📦 PyPI

📝 License

Part of modern-python

Built on modern-di, a dependency-injection framework with an IoC container and scopes.

Browse the full list of templates and libraries in modern-python; the org profile has the categorized index.