Usage with `arq`¶

How to use¶

1. Install `modern-di-arq`¶

uvpippoetry

uv add modern-di-arq

pip install modern-di-arq

poetry add modern-di-arq

2. Apply to your application¶

import dataclasses
import typing

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


@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 render(self) -> str:
        return f"service={self.settings.service_name}"


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


@inject
async def run_report(
    ctx: dict[str, typing.Any],      # arq passes its context dict as the first argument
    report: typing.Annotated[Report, FromDI(Report)],
) -> str:
    return report.render()


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


container = Container(groups=[AppGroup])
setup_di(WorkerSettings, container)
container.validate()  # fails fast on a broken graph before the worker runs

Run the worker as usual — arq mymodule.WorkerSettings — and enqueue jobs from anywhere:

from arq import create_pool
from arq.connections import RedisSettings


async def main() -> None:
    pool = await create_pool(RedisSettings(host="localhost"))
    await pool.enqueue_job("run_report")

setup_di(worker_settings, container) seeds the container into arq's ctx dict (arq's per-worker state store) and wraps four of arq's lifecycle hooks: on_startup/on_shutdown open and close the root container, and on_job_start/on_job_end build and close a Scope.REQUEST child container around each job. Any hook you already defined still runs — yours runs after ours on startup/job-start and before ours on shutdown/job-end, so your code always sees a live container. It accepts a WorkerSettings class (the common case) or a plain settings dict, and returns the container.

@inject resolves each FromDI-annotated parameter from the per-job child container and forwards it to your task. Your task must declare arq's ctx dict as its first parameter (arq calls every task as task(ctx, *args)). Injection is parameter-order-insensitive — a FromDI parameter may sit anywhere in the signature — and a task with no FromDI parameter is returned unchanged.

Scopes¶

The integration builds one Scope.REQUEST child container per job. It is created in on_job_start and closed with close_async() in on_job_end, which arq runs whether the job succeeded or raised — so REQUEST-scoped providers (and their finalizers) live exactly for the duration of one job and never leak on the error path. APP-scoped providers persist for the whole worker: setup_di opens the root container on on_startup and closes it on on_shutdown, running APP-scoped finalizers once when the worker stops.

There is no Scope.SESSION for arq — a job queue has no session concept comparable to a websocket connection.

Async resolution, no connection object¶

FromDI resolves its dependency with Container.resolve_dependency(...), which is synchronous — modern-di's resolution is always sync, regardless of the framework. Container lifecycle here is async, matching arq: the root and each per-job child are closed with close_async(), so REQUEST- and APP-scoped finalizers may be async (or sync).

arq's per-job ctx is a plain dict (job_id, job_try, redis, ...), not a dedicated request/message type, so — like Celery and Typer — modern_di_arq registers no context provider. A task that needs job metadata reads it from the ctx argument arq already passes. If you need the root container elsewhere (for example in your own on_job_start), fetch_di_container(ctx) returns it.

Restart safety¶

setup_di wires container.open() onto on_startup, and calling open() again on an already-open container is a no-op — it unconditionally clears closed and runs no validation, so it costs nothing regardless of graph state. A worker that starts, stops (closing the container), and starts again — a restart, or a test that runs the worker twice — reopens the same container cleanly. Calling setup_di twice on the same worker_settings is rejected with a TypeError, since stacking the hook wrappers would leak a per-job child container.

Tasks with `*args`/`**kwargs`¶

@inject resolves dependencies by binding the task signature by name, which is what makes injection order-insensitive. A task that mixes a FromDI parameter with *args or **kwargs cannot be bound unambiguously, so @inject raises a TypeError at decoration time rather than silently misrouting arguments. Give an @inject task explicit named parameters. (A task with no FromDI parameter is untouched and may use *args/**kwargs freely.)

API¶

Symbol	Description
`setup_di(worker_settings, container)`	Seed the root container into arq's `ctx` and wire root + per-job lifecycle onto arq's `on_startup`/`on_shutdown`/`on_job_start`/`on_job_end` hooks. Accepts a `WorkerSettings` class/object or a settings `dict`; composes with existing hooks; returns the container. Raises `TypeError` if called twice on the same `worker_settings`.
`FromDI(provider_or_type)`	Marker for `Annotated[T, FromDI(...)]` in task signatures; accepts a provider instance or a plain type.
`@inject`	Decorator that resolves `FromDI`-annotated parameters from the per-job `Scope.REQUEST` child container. Order-insensitive; passthrough for tasks with no `FromDI`; raises `TypeError` at decoration if the task also declares `args`/`*kwargs`.
`fetch_di_container(ctx)`	Returns the root container from an arq `ctx` dict.

Usage with arq¶