Celery Integration

Dependency injection for Celery is available in the wireup.integration.celery module.

• Explicit Dependency Injection

  ---

  Task injection is opt-in via @inject.

• Shared Business Logic

  ---

  Wireup is framework-agnostic. Share the same service layer between Celery workers, APIs, and CLIs.

Initialize the integration

First, create a sync container with your dependencies:

from celery import Celery
import wireup

celery_app = Celery("my_app")

container = wireup.create_sync_container(injectables=[services])
wireup.integration.celery.setup(container, celery_app)

Inject in Celery Tasks

Tasks that need Wireup injection must be decorated with @inject.

Celery Task

from wireup import Injected
from wireup.integration.celery import inject


@celery_app.task
@inject
def process_order(order_service: Injected[OrderService], order_id: str) -> None:
    order_service.process(order_id)

Accessing the Container

To access the Wireup container directly, use:

from wireup.integration.celery import get_app_container, get_task_container

# App-wide container created with wireup.create_sync_container
app_container = get_app_container(celery_app)

# Task-scoped container, available only while a task is running
task_container = get_task_container()

API Reference

Visit API Reference for detailed information about the Celery integration module.