modern-di integration for taskiq.
Full guide: taskiq integration docs
Usage example: examples/
uv add modern-di-taskiq # or: pip install modern-di-taskiqsetup_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.
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.
| Symbol | Description |
|---|---|
setup_di(broker, container, *, manage_lifespan=True) |
Stores the root container on broker.state, opens/closes it on worker startup/shutdown, and registers taskiq_message_provider. With manage_lifespan=False it never opens or closes the root container, so the caller owns it; per-task children are built and closed as before. Returns the container |
FromDI(dependency, *, use_cache=True) |
taskiq dependency that resolves a provider instance or a type from the task's Scope.REQUEST child container, building that child on first use; use it as Annotated[T, FromDI(...)] in task signatures. use_cache is passed to TaskiqDepends. Raises RuntimeError naming setup_di when a task reaches it without setup_di called |
fetch_di_container(broker) |
Returns the root container attached to the taskiq broker |
taskiq_message_provider |
ContextProvider for the current taskiq.TaskiqMessage (REQUEST scope) |
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.