modern-di-arq 4.0.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.

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.