Overview

Celery's canvas lets you compose tasks into larger workflows — sequential pipelines, parallel fan-outs, fan-in callbacks, and batched maps — using signatures (.s()) and the primitives chain, group, chord, starmap, map, and chunks. Each composed task still rides KubeMQ Queues under the hood, and because kubemq-celery is a Kombu transport that registers the kubemq:// URL scheme, every canvas primitive works unchanged on KubeMQ. There is no canvas-specific API to learn and no code to rewrite: the same workflow that ran on Redis or RabbitMQ runs on KubeMQ once you switch the broker URL.

Canvas composition is a Celery-level feature, not a broker feature. The transport's job is to deliver task messages and (optionally) store results — chaining, grouping, and callback orchestration all happen in Celery itself.

The only behavioral difference is chord unlocking. Redis provides an O(1) native chord unlock; KubeMQ uses Celery's portable polling fallback (the chord_unlock task polls for group completion). This is covered in detail in the Chord section below.

All examples on this page configure the broker and result backend the same way — point both at a kubemq:// URL:

app.py

import os

from celery import Celery

import kubemq_celery  # noqa: F401 — registers the kubemq:// transport

app = Celery("canvas_app")
app.config_from_object(
    {
        "broker_url": os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
        "result_backend": os.environ.get("CELERY_RESULT_BACKEND", "kubemq://localhost:50000"),
        "result_expires": 3600,
        "task_serializer": "json",
        "result_serializer": "json",
        "accept_content": ["json"],
    }
)

You also need a reachable KubeMQ broker. The fastest way to get one locally is Docker:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Port 50000 is the gRPC port the transport connects to; port 9090 is the shared HTTP server (REST and the /health probe) — curl http://localhost:9090/health confirms the broker is up as tasks flow through your workflow.

import kubemq_celery must run before Celery resolves the broker URL — the import is what registers the kubemq:// scheme with Kombu. Keep it at the top of your app module even if your editor flags it as unused. See Getting Started for the full setup walkthrough.

Chain: sequential pipeline

A chain runs tasks one after another, passing the result of each task as the first argument of the next. Build the pipeline from signatures and call it — the output of add(2, 3) becomes the x argument of multiply, and so on.

chain.py

from celery import chain


@app.task
def add(x: int, y: int) -> int:
    return x + y


@app.task
def multiply(x: int, y: int) -> int:
    return x * y


@app.task
def subtract(x: int, y: int) -> int:
    return x - y


# Pipeline: add(2, 3) -> multiply(result, 10) -> subtract(result, 5)
# Expected: (2 + 3) * 10 - 5 = 45
workflow = chain(
    add.s(2, 3),
    multiply.s(10),
    subtract.s(5),
)

result = workflow.apply_async()
value = result.get(timeout=10)
print(f"Result: {value}")  # 45

Chains can be arbitrarily long. Each step waits for the previous one to complete, and the final AsyncResult resolves to the output of the last task:

chain.py

# add(1,1)=2 -> multiply(2,3)=6 -> add(6,4)=10 -> multiply(10,2)=20
workflow = chain(
    add.s(1, 1),
    multiply.s(3),
    add.s(4),
    multiply.s(2),
)

value = workflow.apply_async().get(timeout=10)
print(f"Result: {value}")  # 20

Group: parallel fan-out

A group dispatches its members in parallel and collects all of their results. Calling .get() on the returned GroupResult blocks until every member completes, then returns the list of results.

group.py

from celery import group


@app.task
def add(x: int, y: int) -> int:
    return x + y


@app.task
def square(n: int) -> int:
    return n * n


# Static group: four additions in parallel
workflow = group(
    add.s(1, 2),
    add.s(3, 4),
    add.s(5, 6),
    add.s(7, 8),
)
values = workflow.apply_async().get(timeout=10)
print(f"Results: {values}")  # [3, 7, 11, 15]

Groups can also be built dynamically from an iterable of signatures — useful when the number of parallel tasks is data-driven:

group.py

# Dynamic group: square each value in 1..5
workflow = group(square.s(i) for i in range(1, 6))
values = workflow.apply_async().get(timeout=10)
print(f"Results: {values}")  # [1, 4, 9, 16, 25]

# Large group: 20 parallel additions
workflow = group(add.s(i, i) for i in range(20))
values = workflow.apply_async().get(timeout=30)
print(len(values))  # 20

Group result collection is backed by the result backend. On KubeMQ, group metadata is stored on celery-group-{group_id} Queue channels, and member results on per-task celery-result-{task_id} channels read back with a non-destructive peek. See Result Backend for how this works.

Chord: group + callback

A chord runs a group of tasks in parallel and then invokes a callback with the list of all group results. It is the canonical fan-out / fan-in pattern: parallel work followed by an aggregation step.

chord.py

from celery import chord, group


@app.task
def add(x: int, y: int) -> int:
    return x + y


@app.task
def sum_results(values: list[int]) -> int:
    """Chord callback — sums the group results."""
    return sum(values)


# Run three additions in parallel, then sum the results.
workflow = chord(
    group(
        add.s(1, 2),
        add.s(3, 4),
        add.s(5, 6),
    ),
    sum_results.s(),
)
value = workflow.apply_async().get(timeout=15)
print(f"sum([3, 7, 11]) = {value}")  # 21

Chord uses the polling fallback on KubeMQ

This is the one place where KubeMQ behaves differently from Redis. Redis provides a native O(1) chord unlock — the broker atomically decrements a counter as each group member completes and fires the callback immediately when the count reaches zero. KubeMQ does not implement that broker-side primitive, so kubemq-celery uses Celery's portable polling fallback: a built-in chord_unlock task periodically polls the result backend for group completion and dispatches the callback once all members have finished.

Chord requires a result backend. The chord_unlock task polls group results to detect completion, so a chord will not unlock unless result_backend is configured. Set result_backend="kubemq://..." (or any supported backend). See Result Backend.

Two practical consequences:

Latency — the callback fires on the next poll after the last member completes, adding roughly 1–2 seconds versus Redis's immediate unlock. Use generous timeouts on result.get() for chord-based workflows.
Correctness is unchanged — the polling fallback is a standard Celery mechanism; results and callback semantics are identical, only the unlock timing differs.

chord.py

# Chord built from a dynamic group — same polling-fallback behavior.
workflow = chord(
    group(add.s(i, i * 2) for i in range(1, 6)),
    sum_results.s(),
)
value = workflow.apply_async().get(timeout=15)
print(f"sum([3, 6, 9, 12, 15]) = {value}")  # 45

Chord error handling

If any chord member fails, the callback is not executed. Attach an error callback with link_error to observe the failure, and propagate the exception to the caller with result.get(..., propagate=True).

chord_error_handling.py

from celery import Task, chord, group


@app.task
def add(x: int, y: int) -> int:
    return x + y


@app.task
def failing_task(x: int) -> int:
    raise ValueError(f"Intentional failure for input {x}")


@app.task
def sum_results(values: list[int]) -> int:
    return sum(values)


@app.task
def on_chord_error(request: Task, exc: Exception, traceback: str | None) -> None:
    """Error callback — runs when a chord member fails."""
    print(f"Chord error! Task {request.id} failed: {exc}")


workflow = chord(
    group(
        add.s(1, 2),
        failing_task.s(99),
        add.s(5, 6),
    ),
    sum_results.s(),
)
workflow.link_error(on_chord_error.s())

try:
    result = workflow.apply_async()
    result.get(timeout=15, propagate=True)
except Exception as exc:
    print(f"Chord failed as expected: {type(exc).__name__}: {exc}")

For broker-level failure handling — dead-letter queues, retries, and acknowledgment semantics — see Error Handling.

Chains of groups and multi-stage workflows

Canvas primitives nest. A common pattern is a chain whose first step is a chord (a parallel stage that collects into a single result), followed by sequential transformation steps. Each stage waits for the previous group to fully complete before the next begins.

chain_of_groups.py

from celery import chain, chord, group


@app.task
def add(x: int, y: int) -> int:
    return x + y


@app.task
def collect(values: list[int]) -> list[int]:
    """Pass-through collector — fans the group results into the chain."""
    return values


@app.task
def double_all(values: list[int]) -> list[int]:
    return [v * 2 for v in values]


@app.task
def sum_all(values: list[int]) -> int:
    return sum(values)


# Stage 1 (chord): parallel additions -> collect           => [3, 7, 11]
# Stage 2 (task):  double every element                    => [6, 14, 22]
# Stage 3 (task):  sum everything                           => 42
workflow = chain(
    chord(
        group(add.s(1, 2), add.s(3, 4), add.s(5, 6)),
        collect.s(),
    ),
    double_all.s(),
    sum_all.s(),
)
value = workflow.apply_async().get(timeout=20)
print(f"Result: {value}")  # 42

The same composition scales to realistic pipelines. The example below models an ETL workflow — extract data from several sources in parallel, transform the merged result through sequential steps, then load it:

complex_workflow.py

import time

from celery import chain, chord, group


@app.task
def fetch_sales_data(region: str) -> dict:
    """Extract — simulate fetching from a regional database."""
    time.sleep(0.3)
    data = {
        "us": {"region": "us", "revenue": 50000, "orders": 120},
        "eu": {"region": "eu", "revenue": 35000, "orders": 85},
        "apac": {"region": "apac", "revenue": 28000, "orders": 65},
    }
    return data.get(region, {"region": region, "revenue": 0, "orders": 0})


@app.task
def merge_datasets(datasets: list[dict]) -> dict:
    """Transform — chord callback that merges the extracted datasets."""
    return {
        "regions": [d["region"] for d in datasets],
        "total_revenue": sum(d["revenue"] for d in datasets),
        "total_orders": sum(d["orders"] for d in datasets),
    }


@app.task
def enrich_summary(summary: dict) -> dict:
    summary["avg_order_value"] = round(
        summary["total_revenue"] / max(summary["total_orders"], 1), 2
    )
    summary["region_count"] = len(summary["regions"])
    return summary


@app.task
def save_report(report: dict) -> dict:
    """Load — simulate writing to a data warehouse."""
    time.sleep(0.2)
    return {"status": "saved", **report}


workflow = chain(
    # Extract: parallel fetch from three regions, merged by the chord callback.
    chord(
        group(
            fetch_sales_data.s("us"),
            fetch_sales_data.s("eu"),
            fetch_sales_data.s("apac"),
        ),
        merge_datasets.s(),
    ),
    # Transform: enrich the merged summary.
    enrich_summary.s(),
    # Load: persist the final report.
    save_report.s(),
)
report = workflow.apply_async().get(timeout=30)
print(report["status"])  # saved

Because the parallel stages use chords, these multi-stage workflows rely on the polling fallback described above — budget for the extra unlock latency at each chord boundary when setting timeouts.

starmap, map, and chunks for batching

When you need to apply one task across a large iterable, starmap, map, and chunks avoid the overhead of dispatching a separate task message per item.

task.starmap(iterable) applies the task to each tuple in the iterable, unpacking each tuple into positional arguments. It runs as a single task that iterates locally:

starmap.py

@app.task
def add(x: int, y: int) -> int:
    return x + y


pairs = [(1, 2), (3, 4), (5, 6), (7, 8)]
values = add.starmap(pairs).apply_async().get(timeout=10)
print(values)  # [3, 7, 11, 15]

task.map(iterable) is the single-argument variant — each item is passed as the sole argument:

map.py

@app.task
def double(n: int) -> int:
    return n * 2


values = double.map([1, 2, 3, 4, 5]).apply_async().get(timeout=10)
print(values)  # [2, 4, 6, 8, 10]

task.chunks(items, n) splits a large iterable into batches of size n, dispatching each chunk as a separate task so the batches can be processed in parallel across workers. The result is a list of per-chunk result lists:

chunks.py

@app.task
def add(x: int, y: int) -> int:
    return x + y


# 10 additions split into chunks of 3.
items = [(i, i + 1) for i in range(10)]
chunked = add.chunks(items, 3).apply_async().get(timeout=15)

# Flatten the per-chunk results.
flat = [v for chunk in chunked for v in chunk]
print(flat)  # [1, 3, 5, 7, 9, 11, 13, 15, 17, 19]

Use starmap/map when per-item work is cheap and you want a single task; use chunks when the dataset is large and you want the batches distributed across workers.

Immutable signatures

By default a chained signature is mutable (.s()): the previous task's result is prepended as its first argument. An immutable signature (.si()) ignores the parent result entirely — its arguments are fixed. Use .si() when a step should run for its side effects, or with a fixed input, regardless of what came before.

immutable_signatures.py

from celery import chain


@app.task
def add(x: int, y: int) -> int:
    return x + y


@app.task
def multiply(x: int, y: int) -> int:
    return x * y


# Mutable: add(2,3)=5 flows into multiply(5, 10) = 50
mutable = chain(add.s(2, 3), multiply.s(10))
print(mutable.apply_async().get(timeout=10))  # 50

# Immutable: add(2,3) runs, but its result is discarded.
# multiply(100, 10) = 1000 — the 5 is ignored.
immutable = chain(add.s(2, 3), multiply.si(100, 10))
print(immutable.apply_async().get(timeout=10))  # 1000

You can mix the two in one chain — mutable steps pass results through, immutable steps reset the argument flow:

immutable_signatures.py

@app.task
def log_and_return(value: int) -> int:
    print(f"  [log_and_return] received: {value}")
    return value


# add(5,5)=10 -> log_and_return(10)=10 -> multiply.si(7,3)=21 (10 discarded)
mixed = chain(
    add.s(5, 5),
    log_and_return.s(),
    multiply.si(7, 3),
)
print(mixed.apply_async().get(timeout=10))  # 21

Result backend requirement

Any canvas pattern that collects or aggregates results depends on the result backend:

group — GroupResult.get() reads each member's stored result.
chord — the chord_unlock task polls the backend to detect group completion before firing the callback.
Chain results — result.get() on the final step reads the stored return value.

Configure the KubeMQ queue-peek backend by pointing result_backend at a kubemq:// URL:

app.py

app.config_from_object(
    {
        "broker_url": "kubemq://localhost:50000",
        "result_backend": "kubemq://localhost:50000",
        "result_expires": 3600,
    }
)

Results are stored as KubeMQ Queue messages on celery-result-{task_id} channels and read back with a non-destructive peek, and group metadata lives on celery-group-{group_id} channels — so no external Redis or database is required.

Without a result backend, group and chord cannot return results and a chord will never unlock. The KubeMQ backend caps result expiry at 24 hours (86400 seconds); Celery's default result_expires of 24 hours already matches this maximum.

For the full backend walkthrough and configuration options, see Result Backend.

Testing canvas in eager mode

You can unit-test canvas composition without a broker or result backend by enabling eager mode, where tasks run synchronously in the calling process. Set task_always_eager=True (and task_eager_propagates=True so errors surface as exceptions), then assert on results.

test_canvas.py

from celery import Celery, chain, chord, group

import kubemq_celery  # noqa: F401

app = Celery("test_canvas")
app.conf.update(
    task_always_eager=True,
    task_eager_propagates=True,
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
)


@app.task
def add(x: int, y: int) -> int:
    return x + y


@app.task
def double(x: int) -> int:
    return x * 2


@app.task
def sum_list(values: list[int]) -> int:
    return sum(values)


def test_chain_basic():
    workflow = chain(add.s(2, 3), double.s())
    result = workflow.apply()
    assert result.result == 10


def test_group_basic():
    workflow = group(add.s(1, 1), add.s(2, 2), add.s(3, 3))
    values = workflow.apply().get()
    assert sorted(values) == [2, 4, 6]


def test_chord_basic():
    workflow = chord(
        group(add.s(1, 1), add.s(2, 2), add.s(3, 3)),
        sum_list.s(),
    )
    result = workflow.apply()
    assert result.result == 12


def test_immutable_signatures():
    # add.si(10, 20) ignores the result of add(1, 2).
    workflow = chain(add.s(1, 2), add.si(10, 20))
    result = workflow.apply()
    assert result.result == 30

Run the tests with pytest:

pytest test_canvas.py -v

Eager mode runs tasks in-process and bypasses the broker entirely, so it validates canvas composition and result-passing logic — not transport behavior. Note that in eager mode chords run synchronously without the chord_unlock polling path; cover the polling-fallback timing against a live KubeMQ broker in an integration test.

Canvas Workflows

Overview

Chain: sequential pipeline

Group: parallel fan-out

Chord: group + callback

Chord uses the polling fallback on KubeMQ

Chord error handling

Chains of groups and multi-stage workflows

starmap, map, and chunks for batching

Immutable signatures

Result backend requirement

Testing canvas in eager mode

Next steps

Scheduling & Delayed Delivery

Result Backend

Error Handling

On this page