tracker

Progress tracking for Celery workloads. Wrap a dispatch in a context manager and every task it publishes is stamped with a shared tracker ID, registered as a member of that tracker, and made queryable — with live Celery state merged in — through a small service and a Django Ninja API.

It answers a question Celery itself does not: "these N tasks belong to one logical job — what's the overall state, and which named steps has it gone through?"

lib/ is on the Python path, so import it as a top-level package:

import tracker
from tracker import track, step, state, configure
from tracker import TrackerBackend, RedisTrackerBackend, DjangoTrackerBackend

Concepts

Concept	What it is
track(title)	Context manager that stamps every task published inside it.
step(title)	Context manager / decorator marking a named phase inside a running task.
state(id)	Load a merged TrackerState (persisted + live Celery) by tracker or task.
TrackerState	The aggregate view: status, progress, member tasks and steps.
ExecutionState	One unit of execution — a member task or a step.
TrackerBackend	Pluggable persistence (Redis or Django ORM).
configure(backend)	Set the process-wide default backend explicitly.

The flow

with track("Nightly ETL") as t:        # 1. persist a TrackerState, open a stamping scope
    extract.delay()                    # 2. before_task_publish stamps each message with t.tracker_id
    transform.delay()                  #    and registers the task id as a member
    load.delay()
                                       # 3. on the worker, tracker.step(...) records named phases
                                       #    against the stamp Celery propagated
state(t.tracker_id)                    # 4. read back merged state (persisted + live AsyncResult)

Members are discovered passively. A before_task_publish signal handler (connected globally on import in tracker.signals, plus a context-local one in track) inspects every outgoing message; if it carries the tracker_id stamp it records the task ID as a member. This fires on the client and on workers (chain/chord callbacks carry the propagated stamp), so the member set is built incrementally with no canvas introspection.

Tracking a dispatch

import tracker

with tracker.track("Q4 Report") as t:
    report_task.delay(report_id=1)

print(t.tracker_id)  # UUID covering everything published in the block

Grouped work shares a single ID:

with tracker.track("Nightly ETL") as t:
    extract.delay()
    transform.delay()
    load.delay()
# t.tracker_id now tracks all three tasks as one job

The TrackerState is persisted on __enter__ (before any task is dispatched) so member registration always finds the row.

Recording steps inside a task

step reads the tracker_id from the stamped header Celery propagates to the worker — no backend lookup or argument plumbing required. Use it as a context manager:

import tracker
from conf.celery import app


@app.task
def my_task():
    with tracker.step("Fetch data"):
        fetch()
    with tracker.step("Transform"):
        transform()

…or as a decorator that wraps the whole body as one step:

@app.task
@tracker.step("Process")
def my_task():
    process()

On exit the step is marked SUCCESS, or FAILURE (capturing the exception) if the block raised — exceptions are never suppressed. Used outside a Celery task, or on a task with no tracker stamp, step logs a warning and becomes a no-op so it is always safe to leave in place.

Reading state

st = tracker.state(tracker_id)   # also accepts a member Celery task id

st.state                 # aggregate state: PENDING / STARTED / SUCCESS / FAILURE / ...
st.progress_completed    # number of member tasks in a terminal state
st.progress_target       # total member tasks
st.tasks                 # {celery_task_id: ExecutionState}
st.steps                 # [ExecutionState] recorded via tracker.step()
st.started_on            # derived from the earliest step / creation time
st.completed_on          # set once the aggregate reaches a terminal state

state() (backed by TrackerService.get_tracker) loads the persisted record, then merges live Celery state for every member via AsyncResult. The per-task states are reduced to one aggregate:

• any FAILURE → FAILURE
• any STARTED/RETRY → STARTED
• all SUCCESS/REVOKED → SUCCESS
• otherwise → PENDING

You can pass either a tracker ID or one of its member Celery task IDs; the service transparently maps a task ID back to its tracker.

Backends

Persistence is pluggable through the TrackerBackend ABC. Two implementations ship:

Redis (default, zero-config)

If you never call configure(), the backend is auto-resolved from the Celery app: an explicit tracker_backend on the app config wins, otherwise a RedisTrackerBackend is built from result_backend when it points at redis:// / rediss://. Tracker state, members, steps and a task→tracker reverse index are stored under celery-tracker:* keys. In this template result_backend already targets Redis, so tracking works out of the box.

Django ORM

DjangoTrackerBackend persists trackers and steps as Django model rows. The model classes are injected so the library stays decoupled from any particular schema:

import tracker
from myapp.models import TrackedTask, TrackedTaskStep

tracker.configure(
    tracker.DjangoTrackerBackend(
        task_model=TrackedTask,
        step_model=TrackedTaskStep,
    )
)

configure() sets a process-wide default; call it once at startup (e.g. in an AppConfig.ready()).

Writing your own

Subclass TrackerBackend and implement save / load / list_all, add_member / get_members, and add_step / update_step / get_steps. add_member must be safe to call concurrently from the client and multiple workers. Optionally override find_tracker_id_for_member to support lookup by member task ID.

HTTP API

tracker.api exposes a Django Ninja router. In this template it is mounted at /tracker (see conf/urls.py):

Method & path	Purpose
GET /tracked-tasks	List all trackers (live state merged).
GET /tracked-tasks/{tracker_id}	Tracker detail with tasks and steps.
GET /tracked-tasks/by-celery-id/{id}	Same detail, keyed by member task ID.
POST /tracked-tasks/{tracker_id}/cancel	Revoke the tracker's Celery tasks.
GET /celery-tasks	List registered Celery task types.
POST /celery-tasks/{task_name}/run	Dispatch a task by name, tracked.

POST /celery-tasks/{task_name}/run wraps the dispatch in track() and returns the new tracker_id, so a frontend can immediately poll /tracked-tasks/{tracker_id} for progress.

Endpoints under /tracked-tasks/{id}/children and /tracked-tasks/stats/* are present for interface compatibility but return 501 Not Implemented.

Testing

For unit tests, configure an in-memory or fake backend explicitly with tracker.configure(...) so tests don't depend on a live Redis or Celery app. Because step no-ops outside a task context, task bodies can be called directly without a worker.

What this is not

• Not a Celery result backend — it sits alongside one and reads live state from it via AsyncResult.
• Not a scheduler or queue — dispatch is still plain .delay() / .apply_async(); track only stamps and observes.
• Not event sourcing — there is no event log; state is a snapshot merged from persisted records and live Celery results.