Usage with `taskiq`¶

How to use¶

1. Install `modern-di-taskiq`¶

uvpippoetry

uv add modern-di-taskiq

pip install modern-di-taskiq

poetry add modern-di-taskiq

2. Apply to your application¶

import dataclasses
import typing

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


@dataclasses.dataclass(kw_only=True, slots=True, frozen=True)
class Settings:
    service_name: str = "catalog"


@dataclasses.dataclass(kw_only=True, slots=True)
class Report:
    settings: Settings   # APP-scoped, injected by type

    def as_dict(self) -> dict[str, str]:
        return {"service": self.settings.service_name}


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


broker = InMemoryBroker()
container = Container(groups=[AppGroup])
setup_di(broker, container)
container.validate()  # after setup_di — its connection providers are now registered


@broker.task
async def get_report(
    report: typing.Annotated[Report, FromDI(Report)],
) -> dict[str, str]:
    return report.as_dict()

setup_di(broker, container) stores the container on broker.state and registers TaskiqEvents.WORKER_STARTUP/WORKER_SHUTDOWN handlers that open/close it — those fire when the broker's worker process starts and stops, so a script that just calls tasks directly (like InMemoryBroker in a test) must drive the container lifecycle itself, e.g. async with broker: ... or an explicit container.open() / await container.close_async().

Deployment: run_receiver_task skips startup by default

taskiq.api.run_receiver_task(...) defaults run_startup=False, which skips the worker startup that opens the root container — tasks still run (the container is already open from construction), but nothing ever closes it, so its finalizers never run at shutdown. Pass run_startup=True (or close the root yourself around consuming) when embedding a receiver with run_receiver_task.

Scopes¶

The integration creates a Scope.REQUEST child container for each task the worker executes. REQUEST-scoped providers (and their finalizers) live for the duration of that one task — the child container is closed after the task returns, including when it raises. APP-scoped providers persist for the whole worker process; setup_di opens the APP container on WORKER_STARTUP and runs await container.close_async() on WORKER_SHUTDOWN.

There is no Scope.SESSION for taskiq — a task queue doesn't have a session concept comparable to websockets.

Sync resolution, async cleanup¶

FromDI resolves its dependency with Container.resolve_dependency(...), which is synchronous — modern-di's resolution is always sync, regardless of the framework. The per-task Scope.REQUEST child container that resolution runs against is nevertheless torn down asynchronously: after the task handler finishes (or raises), the integration awaits container.close_async() on it. So async finalizers on REQUEST-scoped providers run correctly, while the factories themselves must build synchronously.

Framework context objects¶

taskiq.TaskiqMessage is automatically made available by the integration, so factories can declare it as a parameter and get the message that triggered the current task — see Framework Context Objects for how implicit and explicit resolution work.

The following context provider is also available for explicit import:

taskiq_message_provider — provides the current taskiq.TaskiqMessage object.

Implicit (type-based) usage¶

import taskiq
from modern_di import Group, Scope, providers


def create_task_info(message: taskiq.TaskiqMessage) -> dict[str, str]:
    return {
        "task_id": message.task_id,
        "task_name": message.task_name,
    }


class AppGroup(Group):
    # The message dependency is resolved by type annotation
    task_info = providers.Factory(
        create_task_info,
        scope=Scope.REQUEST,
    )

Explicit (provider-based) usage¶

import taskiq
import modern_di_taskiq
from modern_di import Group, Scope, providers


def create_task_info(message: taskiq.TaskiqMessage) -> dict[str, str]:
    return {"task_id": message.task_id}


class AppGroup(Group):
    task_info = providers.Factory(
        create_task_info,
        scope=Scope.REQUEST,
        kwargs={"message": modern_di_taskiq.taskiq_message_provider},
    )

API¶

Symbol	Description
`setup_di(broker, container)`	Wire the APP-scope container into taskiq — creates a REQUEST child container per task and opens/closes the APP container on worker startup/shutdown.
`FromDI(provider_or_type)`	Marker for `Annotated[T, FromDI(...)]` in task signatures; accepts a provider instance or a plain type.
`fetch_di_container(broker)`	Returns the APP-scope container registered with the taskiq broker.
`taskiq_message_provider`	`ContextProvider` for the current `taskiq.TaskiqMessage`.

Usage with taskiq¶