Celery propagation

When Falcon request-handling code publishes Celery tasks, the current correlation ID can be copied into the outgoing task message's Advanced Message Queuing Protocol (AMQP) correlation_id property. When a Celery worker later executes a task carrying that value, falcon-correlate also makes it available through correlation_id_var for the lifetime of the task body.

Note: Celery is an optional dependency. Install the package with the Celery extra in any process that publishes tasks:

pip install "falcon-correlate[celery]"

If the Celery extra is not installed, importing falcon_correlate and falcon_correlate.celery remains safe. The Celery signal integration simply stays inactive, and Celery-specific project tests are reported as skipped in test environments where the optional dependency is absent.

The clearest activation path is to call configure_celery_correlation(celery_app) during Celery application setup in each publisher and worker process:

from celery import Celery
from falcon_correlate import CorrelationIDMiddleware, configure_celery_correlation

celery_app = configure_celery_correlation(
    Celery("myapp", broker="redis://localhost:6379/0")
)
middleware = CorrelationIDMiddleware()

The helper connects the before_task_publish, task_prerun, and task_postrun handlers in one idempotent call and returns the same app instance. It is safe to call repeatedly, which is useful when publisher and worker bootstrap code share an application factory.

Importing falcon_correlate also registers those handlers automatically when Celery is installed. If the publisher process or worker process already imports anything from the package root, no extra registration call is needed. The explicit helper remains the preferred form for new application setup because it makes Celery integration visible at the point where the Celery app is created.

Once the handlers are registered in the publisher process, normal task publishing APIs such as delay() and apply_async() propagate the request correlation ID automatically. When the worker process also registers the handlers, task execution sees that incoming value through correlation_id_var.get().

from falcon_correlate import correlation_id_var


def enqueue_invoice_email(user_id: str) -> None:
    # During request handling, CorrelationIDMiddleware has already populated
    # correlation_id_var for the current context.
    send_invoice_email.delay(user_id)
    assert correlation_id_var.get() is not None

from falcon_correlate import correlation_id_var


@celery_app.task(bind=True)
def send_invoice_email(self, user_id: str) -> None:
    task_correlation_id = correlation_id_var.get()
    assert task_correlation_id == self.request.correlation_id

• When correlation_id_var is set and the active Celery result backend is not rpc://, the outgoing Celery message uses that value as its publish-time correlation_id.
• When correlation_id_var is not set, falcon-correlate leaves Celery's generated publish value unchanged.
• If the caller passes apply_async(correlation_id=...) while a request correlation ID is present, the ambient request value wins. Celery already fills correlation_id by default, so overwriting the publish value is the only way to guarantee request-to-task propagation through before_task_publish.
• If the active Celery result backend is rpc://, falcon-correlate preserves Celery's task-id correlation contract and does not overwrite the publish correlation_id.
• Inside a running Celery task, correlation_id_var.get() returns the incoming task request correlation ID when one is present.
• After the task finishes, falcon-correlate resets worker context automatically so the next task does not inherit stale correlation data.