modern-di-celery 4.1.0

modern-di integration for Celery 5.x.

Full guide: Celery integration docs

Usage example: examples/

Installation

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

Usage

setup_di registers the container for the app and connects handlers that open and close it on worker_process_init/worker_process_shutdown (once per forked process under the prefork and solo pools) and on worker_init/worker_shutdown (once in the main worker process, the only signals the threads, gevent, and eventlet pools send). @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")
container = Container(groups=[AppGroup])
setup_di(app, container)
container.validate()  # optional fail-fast check of the provider graph


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

These worker signals fire only when a real celery worker starts and stops, so a script or test that runs tasks eagerly (task_always_eager = True) must drive the container lifecycle itself, either by sending those signals or by opening and closing 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.

Prefork with spawn

Celery runs its prefork pool on billiard, its own copy of multiprocessing, which starts children with the fork method by default on macOS and Linux. With spawn (Windows, or billiard.set_start_method("spawn", force=True)), each child unpickles the app and re-imports your task modules, so setup_di runs again in the child. The worker_process_init handler then registers the child's container for the app the child runs. It matches apps by main name, so give each app in a process its own name. Under spawn, Celery's prefork pool also needs FORKED_BY_MULTIPROCESSING=1 in the environment. Without it tasks fail inside Celery's tracer, with or without modern-di.

Sharing a container

To share one container between the Celery app and other code in the same process, such as another app built on modern-di, pass manage_lifespan=False and open and close the root container yourself. The worker signals then leave it alone, and each task call still gets its own child container:

from modern_di_celery import setup_di


setup_di(app, container, manage_lifespan=False)


async def main() -> None:
    async with container:
        ...  # run the apps that share the container

Under the prefork pool each forked child inherits the open container. With manage_lifespan=False nothing closes it in the child, so close it yourself if its Scope.APP finalizers must run there, for example from your own worker_process_shutdown handler that calls container.close_sync().

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.