TopolearnMulti-step pipelines (fetch, process, aggregate) need to pass results between tasks without blocking workers.
A signature wraps a task call so it can be passed around: add.s(2, 2) (short for add.signature((2, 2))). Signatures can be partial; when used in a workflow, the previous task's result is prepended to the arguments. add.si(2, 2) is immutable: it ignores the parent's result.
The canvas primitives:
- chain:
chain(fetch.s(url), parse.s(), store.s())or with the pipe,fetch.s(url) | parse.s() | store.s(). Each task's return value becomes the first argument of the next. - group:
group(resize.s(img) for img in images)runs tasks in parallel and returns aGroupResult. - chord:
chord(header_group)(callback.s())runs the group, then calls the callback with the list of all results. The classic map-reduce. It needs a result backend.
Call .delay() or .apply_async() on the whole workflow. Error handling attaches with link_error=handler.s() (an errback). In a chain, a failing task stops the rest.
Workflows replace the anti-pattern of calling result.get() inside a task. Keep payloads small, because each step's result passes through the backend and broker. Chaining a group into another task automatically upgrades it to a chord.
Example
from celery import Celery, chain, group, chord
app = Celery("tasks", broker="memory://", backend="cache+memory://")
app.conf.task_always_eager = True # run inline for this demo
@app.task
def add(x, y): return x + y
@app.task
def mul(x, y): return x * y
@app.task
def total(numbers): return sum(numbers)
print(chain(add.s(2, 3), mul.s(10)).delay().get()) # 50: (2+3)*10
print((add.s(1, 1) | add.s(5) | mul.s(2)).delay().get()) # 14
print(group(add.s(i, i) for i in range(4)).delay().get()) # [0, 2, 4, 6]
print(chord(add.s(i, i) for i in range(4))(total.s()).get()) # 12
print(add.si(1, 1).id is None) # True: not sent yetRemember
- task.s(...) makes a signature; .si(...) ignores parent results
- chain passes each result to the next; group runs in parallel
- chord = group + callback receiving all results; needs a result backend
- Use workflows instead of result.get() inside tasks
Watch out for
- Using .s() where the parent result shouldn't be passed, producing a 'takes 2 positional arguments but 3 were given' error.