Workflow Canvas

Workflow Canvas

Purpose: Implement Celery canvas patterns for composing complex, distributed task workflows.

Activation Triggers:

• Sequential task execution needed (chains)
• Parallel task processing required (groups)
• Synchronization after parallel work (chords)
• Complex workflow composition
• Task result aggregation
• Error handling in workflows
• Nested or conditional workflows

Key Resources:

• scripts/test-workflow.sh - Test workflow execution and validate patterns
• scripts/validate-canvas.sh - Validate canvas structure and dependencies
• scripts/generate-workflow.sh - Generate workflows from templates
• templates/ - Complete workflow pattern implementations
• examples/ - Real-world workflow scenarios with explanations

Canvas Primitives

1. Signatures

Foundation of canvas system - encapsulate task invocation details:

from celery import signature

# Named signature
signature('tasks.add', args=(2, 2), countdown=10)

# Using task method
add.signature((2, 2), countdown=10)

# Shortcut syntax (most common)
add.s(2, 2)

# Immutable signature (prevents result forwarding)
add.si(2, 2)

Use templates/chain-workflow.py for signature examples.

2. Chains

Sequential task execution where each task's result becomes next task's first argument:

from celery import chain

# Explicit chain
chain(add.s(2, 2), add.s(4), add.s(8))()

# Pipe syntax (recommended)
(add.s(2, 2) | add.s(4) | add.s(8))()

# With immutable tasks (independent execution)
(add.si(2, 2) | process.s() | notify.si('done'))()

Key Pattern: Use .si() when task should NOT receive previous result.

See templates/chain-workflow.py for complete patterns.

3. Groups

Execute multiple tasks in parallel, returns GroupResult:

from celery import group

# Parallel execution
group(add.s(i, i) for i in range(10))()

# With result tracking
job = group(process.s(item) for item in batch)
result = job.apply_async()
result.get()  # Wait for all tasks

Requirements:

• Tasks must NOT ignore results
• Result backend required

See templates/group-parallel.py for implementation.

4. Chords

Group header + callback that executes after all header tasks complete:

from celery import chord

# Basic chord
chord(add.s(i, i) for i in range(100))(tsum.s()).get()

# With error handling
chord(
    process.s(item) for item in batch
)(aggregate_results.s()).on_error(handle_chord_error.s())

Critical Requirements:

• Result backend MUST be enabled
• Set task_ignore_result=False explicitly
• Redis 2.2+ for proper operation

Error Handling: Failed header tasks → callback receives ChordError with task ID and exception.

See templates/chord-pattern.py for complete implementation.

5. Map & Starmap

Built-in tasks for sequence processing (single message, sequential execution):

# Map: one call per element
add.map([(2, 2), (4, 4), (8, 8)])

# Starmap: unpacks tuples
add.starmap(zip(range(100), range(100)))

Difference from groups: Single task message vs multiple messages.

6. Chunks

Divide large iterables into sized batches:

# Process 100 items in chunks of 10
add.chunks(zip(range(100), range(100)), 10)

Returns: Nested lists corresponding to chunk outputs.

Advanced Patterns

Partial Application

# Incomplete signature
partial = add.s(2)

# Complete later (arguments prepended)
partial.delay(4)  # Executes add(4, 2)

# Kwargs merge with precedence
partial = process.s(timeout=30)
partial.apply_async(kwargs={'timeout': 60})  # Uses 60

Nested Workflows

Combine primitives for complex logic:

# Chain of chords
workflow = chain(
    chord([task1.s(), task2.s()])(callback1.s()),
    chord([task3.s(), task4.s()])(callback2.s())
)

# Groups in chains
workflow = (
    prepare.s() |
    group([process.s(i) for i in range(10)]) |
    finalize.s()
)

See templates/nested-workflows.py for production patterns.

Error Callbacks

# Task-level error handling
add.s(2, 2).on_error(log_error.s()).delay()

# Chord error handling
chord(
    header_tasks
)(callback.s()).on_error(handle_error.s())

Errback signature: (request, exc, traceback)

Execution: Synchronous in worker.

See templates/error-handling-workflow.py for comprehensive patterns.

Complex Workflow Example

from celery import chain, group, chord

# Data processing pipeline
workflow = chain(
    # Stage 1: Fetch and validate
    fetch_data.s(),
    validate_data.s(),

    # Stage 2: Parallel processing
    group([
        transform_batch.s(i) for i in range(num_batches)
    ]),

    # Stage 3: Aggregate results
    chord([
        aggregate_batch.s(i) for i in range(num_batches)
    ])(finalize_report.s()),

    # Stage 4: Notify
    send_notification.si('pipeline_complete')
)

# Execute with error handling
result = workflow.apply_async(
    link_error=handle_pipeline_error.s()
)

See templates/complex-workflow.py for production-ready implementation.

Testing Workflows

Validate Canvas Structure

# Check workflow composition
./scripts/validate-canvas.sh path/to/workflow.py

# Validates:
# - Result backend enabled
# - Task ignore_result settings
# - Proper signature usage
# - Error callback patterns

Test Workflow Execution

# Run workflow with test data
./scripts/test-workflow.sh workflow_name

# Options:
# --dry-run    : Validate without execution
# --verbose    : Show detailed task flow
# --timeout 30 : Set execution timeout

Best Practices

DO:

• Enable result backend for groups/chords
• Use .si() for independent tasks in chains
• Implement error callbacks for critical workflows
• Set explicit timeouts for long-running workflows
• Use chunks for large data processing
• Add metadata with stamping API (5.3+)
• Make error callbacks idempotent

DON'T:

• Have tasks wait synchronously for other tasks
• Ignore results for tasks in groups
• Forget to call super() in after_return() overrides
• Use chords without Redis 2.2+
• Create deeply nested workflows (>3 levels)
• Mix synchronous and async task calls

Configuration Requirements

Celery Config:

# Required for canvas
result_backend = 'redis://localhost:6379/0'
result_extended = True  # Store task args/kwargs

# Recommended
task_track_started = True
result_expires = 3600  # 1 hour

# For large workflows
worker_prefetch_multiplier = 1
task_acks_late = True

Debugging

Visualize workflow:

from celery import DependencyGraph

graph = workflow.__graph__()
with open('workflow.dot', 'w') as f:
    f.write(graph.to_dot())

# Convert to image:
# dot -Tpng workflow.dot -o workflow.png

Monitor execution:

result = workflow.apply_async()
print(f"Task ID: {result.id}")
print(f"Status: {result.status}")
print(f"Children: {result.children}")

Resources

Templates:

• chain-workflow.py - Sequential task patterns
• group-parallel.py - Parallel execution patterns
• chord-pattern.py - Synchronization patterns
• complex-workflow.py - Multi-stage pipelines
• error-handling-workflow.py - Error callback patterns
• nested-workflows.py - Advanced composition

Scripts:

• test-workflow.sh - Execute and validate workflows
• validate-canvas.sh - Static analysis of canvas patterns
• generate-workflow.sh - Generate workflows from templates

Examples:

• examples/chain-example.md - Real-world chain scenarios
• examples/group-example.md - Parallel processing use cases
• examples/chord-example.md - Synchronization patterns
• examples/complex-workflows.md - Production workflow architectures

Security Compliance

This skill follows strict security rules:

• All code examples use placeholder values only
• No real API keys, passwords, or secrets
• Environment variable references in all code
• .gitignore protection documented

Version: 1.0.0 Celery Compatibility: 5.0+ Required Backend: Redis 2.2+ or compatible result backend