modern-di-arq 4.1.0

modern-di integration for arq.

Full guide: arq integration docs

Usage example: examples/

Installation

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

Usage

setup_di seeds the root container into arq's ctx dict and wires four of arq's lifecycle hooks: on_startup/on_shutdown open and close the root container, on_job_start builds a Scope.REQUEST child container per job, and on_job_end closes it as a fallback.

Decorate a task with @inject to resolve its FromDI-marked parameters from that per-job child. The child is open from on_job_start until the last @inject-decorated call for the job exits, which closes it. Nested and concurrent (asyncio.gather) @inject calls over the same job share the one child through a reference count, so it closes exactly once, and teardown still happens if arq skips on_job_end.

import typing

from arq.connections import RedisSettings
from modern_di import Container, Group, Scope, providers
from modern_di_arq import FromDI, inject, setup_di


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)


@inject
async def greet(
    ctx: dict[str, typing.Any],  # arq passes its context dict as the first argument
    name: str,
    greeter: typing.Annotated[Greeter, FromDI(Greeter)],  # resolve by type
) -> str:
    return greeter.greet(name)


class WorkerSettings:
    functions = [greet]
    redis_settings = RedisSettings(host="localhost")


container = Container(groups=[AppGroup])
setup_di(WorkerSettings, container)
container.validate()  # optional fail-fast check of the provider graph

Run the worker as usual (arq mymodule.WorkerSettings) and enqueue jobs with only their real arguments, as in await pool.enqueue_job("greet", "world"). The FromDI parameters are resolved for you. A task must declare arq's ctx dict as its first parameter; injection is order-insensitive otherwise. arq's ctx is a plain dict (not a dedicated message type), so no context provider is registered. Read job metadata from ctx, and fetch_di_container(ctx) returns the root container.

Sharing a container

When one container serves several apps or workers, only one owner should open and close the root. Otherwise the first worker to shut down closes it for everyone else. Pass manage_lifespan=False and own the root yourself:

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

async with container:
    ...  # run the worker and the other apps here

The worker still builds and closes a child container per job, and your own on_startup and on_shutdown hooks still run. 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.