modern-di-taskiq

Modern-DI integration for taskiq.

Quickstart

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 Dependencies(Group):
    settings = providers.Factory(scope=Scope.APP, creator=Settings)


broker = InMemoryBroker()
setup_di(broker, Container(groups=[Dependencies], validate=True))


@broker.task
async def greet(name: str, settings: typing.Annotated[Settings, FromDI(Dependencies.settings)]) -> str:
    return f"{settings.greeting}, {name}"

See the documentation for the full guide.