Skip to content
z4j docs

Celery

Requires Celery 5.2.2+, Python 3.11+. See the compatibility matrix for the full pin string.

z4j is a Celery dashboard, monitoring, and control plane that captures task lifecycle events, persists them in Postgres with a tamper-evident audit chain, and lets operators retry, cancel, or schedule tasks from a unified UI. See the marketing landing for the Celery feature overview and the Flower alternative page if you're migrating from Flower.

Install

Section titled “Install”
Terminal window
pip install z4j-celery

Pip transitively pulls z4j-core and z4j-bare. No explicit adapter registration is needed if you use z4j-django / z4j-flask / z4j-fastapi. The worker_ready signal handler is wired:

  • by z4j-django (which eagerly imports z4j_celery at module-load time when in INSTALLED_APPS), or
  • when a Flask or FastAPI application imports its z4j framework package and the Celery worker imports that application module, or
  • by direct import z4j_celery in your Celery application module.

Celery does not discover this package through a Celery plugin entry point. Ensure the worker process imports one of the paths above.

Worker bootstrap

Section titled “Worker bootstrap”

When a Celery worker is ready, the worker_ready signal fires and z4j's worker bootstrap:

  1. Confirms the process is celery worker (not celery inspect/control/purge/etc. - those don't deserve an agent slot).
  2. Resolves the Celery app from sender.app or current_app.
  3. Calls install_agent(engines=[CeleryEngineAdapter(celery_app=...)]), which reads Z4J_BRAIN_URL / Z4J_TOKEN / Z4J_HMAC_SECRET / Z4J_PROJECT_ID / Z4J_AGENT_NAME from the environment.
  4. Logs INFO:z4j.celery.worker_bootstrap:z4j worker bootstrap: agent runtime started (celery_app=..., framework=...) on success.

If you don't see that log line on worker boot, the agent is not running and tasks will not reach z4j. Check the Z4J_* env vars and that the worker is actually invoked as celery worker (the bootstrap signal only fires under that command).

What it captures

Section titled “What it captures”
Signal z4j event
task_received task.received
task_prerun task.started
task_postrun (state=SUCCESS) task.succeeded
task_postrun (state=FAILURE) task.failed
task_retry task.retried
task_revoked task.revoked

Plus the payload: args, kwargs (redacted), queue, routing key, exchange, retries count, ETA, expires, task parent (chord / group support).

Actions

Section titled “Actions”
Verb How z4j performs it
retry app.send_task(...); reuses result-backend inputs when result_extended stored them, or requires complete operator replacements
cancel fire-and-forget app.control.revoke(task_id, terminate=True, signal="SIGTERM")
purge_queue app.control.purge() on the queue
bulk_retry loop native retry, capped at 10k

Chord / group support

Section titled “Chord / group support”

Celery's chord and group primitives carry a group_id. z4j preserves this, letting the dashboard:

  • Show sub-tasks grouped under the parent.
  • Attribute failures to the root group in alerts.

Caveats

Section titled “Caveats”
  • Canvas signatures - .si() / .s() are not reconstructed as signature objects. Retry uses complete unredacted inputs from the result backend when available, or complete operator-supplied replacements; a redacted event payload is not replay authority.
  • eta/countdown on retry - the original ETA is not recovered automatically; an operator-supplied retry ETA is honored.
  • Broker state visibility - queue length reporting requires redis broker; RabbitMQ support works but depends on rabbitmq-management plugin.

Config

Section titled “Config”

The adapter takes no explicit config - it reads the Celery app's config via the framework adapter.

For Django, the app is auto-detected via 5 candidates (see quickstart §Auto-detect). If your layout is unusual:

# settings.py - top-level Django setting, NOT inside the Z4J dict
CELERY_APP = "myproject.celery:app"

For Flask / FastAPI / bare-Python layouts, pass the app directly to the engine:

from z4j_celery.engine import CeleryEngineAdapter
from z4j_bare.install import install_agent


install_agent(
    engines=[CeleryEngineAdapter(celery_app=my_celery_app)],
    # brain_url / token / hmac_secret / project_id / agent_name
    # are read from Z4J_* env vars if not passed explicitly
)

Performance notes

Section titled “Performance notes”
  • Signal handlers only map and queue events locally; they make no brain network request inline.

See scheduler: celery-beat.