Canvas

Workflow primitives — Signature, chain, group, chord, chunks, starmap.

Canvas primitives for composing task workflows. Import directly from the package:

python

from flexiq import chain, group, chord, chunks, starmap

apply_async() means something different here

On Signature, chain, group, and chord, apply_async() is an async def coroutine — you must await it. This is the opposite of TaskWrapper.apply_async() (the method on a plain @queue.task()-decorated function), which is synchronous and returns a JobResult directly.

Canvas primitives are in-thread orchestrators: chain.apply() / chord.apply() enqueue each step and block on its result in the calling thread, driving the whole pipeline client-side. Unlike Celery, there is no broker-linked callback chain running server-side — the state lives only in the process that called apply().

Signature

A frozen task call spec — describes what to call and with what arguments, without executing it.

Creating signatures

python

# Mutable signature — receives previous result in chains
sig = add.s(2, 3)

# Immutable signature — ignores previous result in chains
sig = add.si(2, 3)

Fields

Field	Type	Description
`task`	`TaskWrapper`	The task to call
`args`	`tuple`	Positional arguments
`kwargs`	`dict`	Keyword arguments
`options`	`dict`	Enqueue options (priority, queue, etc.)
`immutable`	`bool`	If `True`, ignores previous result in chains

`sig.apply()`

python

sig.apply(queue: Queue | None = None) -> JobResult

Enqueue this signature immediately. If queue is None, uses the task's parent queue.

python

sig = add.s(2, 3)
job = sig.apply()
print(job.result(timeout=10))  # 5

`await sig.apply_async()`

python

await sig.apply_async(queue: Queue | None = None) -> JobResult

Async version of apply(). Safe to call from async contexts (FastAPI handlers, etc.).

Mutable vs immutable

In a chain, the previous task's return value is prepended to a mutable signature's args:

python

# add.s(10) in a chain where previous step returned 5:
# → add(5, 10) = 15

# add.si(2, 3) in a chain:
# → add(2, 3) = 5  (always, regardless of previous result)

chain

Execute signatures sequentially, piping each result to the next.

Constructor

python

chain(*signatures: Signature)

Requires at least one signature.

`chain.apply()`

python

chain.apply(queue: Queue | None = None) -> JobResult

Execute the chain by enqueuing and waiting for each step sequentially. Returns the JobResult of the last step.

Each step's return value is prepended to the next mutable signature's args. Immutable signatures (task.si()) receive their args as-is.

python

@queue.task()
def double(x):
    return x * 2

@queue.task()
def add_ten(x):
    return x + 10

# double(3) → 6, then add_ten(6) → 16
result = chain(double.s(3), add_ten.s()).apply()
print(result.result(timeout=10))  # 16

`await chain.apply_async()`

python

await chain.apply_async(queue: Queue | None = None) -> JobResult

Async version of apply(). Awaits each step's result using aresult() instead of blocking with result(). Safe to call from async contexts.

python

result = await chain(double.s(3), add_ten.s()).apply_async()
value = await result.aresult(timeout=10)  # 16

group

Execute signatures in parallel and collect all results.

Constructor

python

group(*signatures: Signature)

Requires at least one signature.

`group.apply()`

python

group.apply(queue: Queue | None = None) -> list[JobResult]

Enqueue all signatures and return a list of JobResult handles. Jobs run concurrently across available workers.

python

jobs = group(
    add.s(1, 2),
    add.s(3, 4),
    add.s(5, 6),
).apply()

results = [j.result(timeout=10) for j in jobs]
print(results)  # [3, 7, 11]

`await group.apply_async()`

python

await group.apply_async(queue: Queue | None = None) -> list[JobResult]

Async version of apply(). With max_concurrency, uses asyncio.gather to await each wave concurrently instead of blocking.

chord

Run a group in parallel, then pass all results to a callback.

Constructor

python

chord(group_: group, callback: Signature)

Parameter	Type	Description
`group_`	`group`	The group of tasks to run in parallel
`callback`	`Signature`	The task to call with all collected results

`chord.apply()`

python

chord.apply(queue: Queue | None = None) -> JobResult

Execute the group, wait for all results, then run the callback with the list of results prepended to its args (unless immutable). Returns the JobResult of the callback.

python

@queue.task()
def fetch(url):
    return requests.get(url).text

@queue.task()
def merge(results):
    return "\n".join(results)

result = chord(
    group(fetch.s("https://a.com"), fetch.s("https://b.com")),
    merge.s(),
).apply()

combined = result.result(timeout=30)

`await chord.apply_async()`

python

await chord.apply_async(queue: Queue | None = None) -> JobResult

Async version of apply(). Awaits all group results using asyncio.gather, then enqueues the callback.

chunks

Split an iterable into fixed-size chunks and process each chunk in parallel.

Constructor

python

chunks(task: TaskWrapper, items: list, chunk_size: int) -> group

Parameter	Type	Description
`task`	`TaskWrapper`	The task to call for each chunk. Receives the chunk as a single positional argument.
`items`	`list`	The full list to split. Must be non-empty.
`chunk_size`	`int`	Items per chunk. Must be positive.

Returns a group of signatures — one per chunk. Apply it the same way as any other group.

python

@queue.task()
def process_batch(records):
    return [transform(r) for r in records]

records = load_records()  # 10_000 items
jobs = chunks(process_batch, records, 100).apply()  # → 100 parallel jobs

results = [j.result(timeout=60) for j in jobs]

Raises ValueError if chunk_size <= 0 or items is empty.

starmap

Spread an iterable of argument tuples over parallel task invocations.

Constructor

python

starmap(task: TaskWrapper, args_list: list[tuple]) -> group

Parameter	Type	Description
`task`	`TaskWrapper`	The task to call. Each tuple is unpacked into positional args.
`args_list`	`list[tuple]`	One tuple per invocation. Must be non-empty.

Returns a group of signatures — one per tuple. Equivalent to group(task.s(*a) for a in args_list).

python

@queue.task()
def add(x, y):
    return x + y

jobs = starmap(add, [(1, 2), (3, 4), (5, 6)]).apply()
results = [j.result(timeout=10) for j in jobs]  # [3, 7, 11]

Raises ValueError if args_list is empty.

Complete example

An ETL pipeline using all three primitives:

python

from flexiq import Queue, chain, group, chord

queue = Queue()

@queue.task()
def extract(source):
    return load_data(source)

@queue.task()
def transform(data):
    return clean(data)

@queue.task()
def aggregate(results):
    return merge_datasets(results)

@queue.task()
def load(data):
    save_to_warehouse(data)

# Extract from 3 sources in parallel, transform each,
# aggregate all results, then load
pipeline = chain(
    chord(
        group(
            chain(extract.s("db"), transform.s()),
            chain(extract.s("api"), transform.s()),
            chain(extract.s("csv"), transform.s()),
        ),
        aggregate.s(),
    ),
    load.s(),
)

result = pipeline.apply(queue)