modern-di-celery

Modern-DI integration for Celery 5.x.

Full guide: Celery integration docs

Installation

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

Usage

setup_di stores the container on app.conf and registers worker_process_init/worker_process_shutdown handlers that open/close it once per worker process. @inject builds a Scope.REQUEST child container per task call and resolves the FromDI-marked parameters from it; alternatively set task_cls=DITask to inject every task without a per-task decorator.

import typing

from celery import Celery
from modern_di import Container, Group, Scope, providers
from modern_di_celery 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)


app = Celery("myapp", broker="redis://localhost")
setup_di(app, Container(groups=[AppGroup], validate=True))


@app.task
@inject
def greet(
    name: str,
    greeter: typing.Annotated[Greeter, FromDI(Greeter)],   # resolve by type
) -> str:
    return greeter.greet(name)

The worker_process_init/worker_process_shutdown signals fire only when a real celery worker process starts and stops, so a script or test that runs tasks eagerly (task_always_eager = True) must drive the container lifecycle itself — send those signals or open/close the container directly. Celery tasks are sync, so REQUEST-scoped finalizers run via close_sync(); there is no framework request/message object, so no context provider is registered.

API

Symbol	Description
setup_di(app, container)	Stores the APP-scope container on app.conf and opens/closes it on worker_process_init/worker_process_shutdown. Returns the container
FromDI(dependency)	Inert marker for Annotated[T, FromDI(...)] in task signatures; accepts a provider instance or a type
inject(task)	Decorator that builds a Scope.REQUEST child per call, resolves the FromDI-annotated parameters, and closes the child with close_sync() afterwards
DITask	Task subclass that applies @inject to a task's run automatically; pass task_cls=DITask to Celery(...) or base=DITask to @app.task(...)
fetch_di_container(app)	Returns the APP-scope container registered with the Celery app

📦 PyPI

📝 License

Part of modern-python

Built on modern-di, a dependency-injection framework with IoC container and scopes.

Browse the full list of templates and libraries in modern-python — see the org profile for the categorized index.