Celery Subtasks: Task Signatures, Partial Arguments, and Composition

In this tutorial, you will learn about Celery Subtasks: Task Signatures, Partial Arguments, and Composition. We cover key concepts, practical examples, and best practices to help you master this topic.

Celery subtasks (task signatures) are pre-configured task representations that bind arguments and options without executing the task, enabling dynamic workflow construction, partial application, and reusable task templates.

flowchart LR
    Task[add(x, y)] --> Signature[Signature: add.s(10)]
    Signature --> Partial[Partial: add(10, ?)]
    Partial --> Chain[Chain: add.s(10) | multiply.s(2)]
    Partial --> Group[Group: [add.s(1), add.s(2)]]
    Signature --> Immutable[Immutable: add.si(10, 20)]
    Immutable --> Fixed[Fixed: add(10, 20)]

What You'll Learn

• Creating task signatures with .s() and .signature()
• Partial arguments and argument merging in chains
• Immutable subtasks (.si()) for fixed arguments
• Cloning, combining, and inspecting signatures

Why It Matters

Task signatures decouple task definition from task execution. They let you pass tasks as arguments to other tasks, build workflows dynamically at runtime, and reuse task configurations without code duplication. Without signatures, canvas workflows would not be possible.

Real-World Use

DodaTech's notification system uses task signatures as templates. A send_email.s() signature is created with the email template and stored in a database. When an event triggers the notification, the remaining arguments (user email, personalization data) are bound to the signature and executed.

Creating Task Signatures

Create and inspect task signatures:

from celery import Celery

app = Celery('signatures', broker='redis://localhost:6379/0')

@app.task
def add(x, y):
    print(f"add({x}, {y}) = {x + y}")
    return x + y

@app.task
def multiply(x, y):
    print(f"multiply({x}, {y}) = {x * y}")
    return x * y

sig = add.s(10, 20)
print(f"Signature: {sig}")
print(f"Task: {sig.task}")
print(f"Args: {sig.args}")
print(f"Kwargs: {sig.kwargs}")
print(f"Options: {sig.options}")

result = sig()
print(f"Direct execution: {result}")

sig_with_kwargs = add.s(x=5, y=15)
result = sig_with_kwargs()
print(f"Keyword args: {result}")

Expected output:

Signature: canvas.add(10, 20)
Task: canvas.add
Args: (10, 20)
Kwargs: {}
Options: {}
Direct execution: 30
Keyword args: 20

Partial Arguments in Chains

Bind some arguments now, others later:

from celery import Celery, chain

app = Celery('signatures', broker='redis://localhost:6379/0')

@app.task
def process_order(item_id, quantity, discount=0):
    subtotal = quantity * 100
    total = subtotal * (1 - discount / 100)
    print(f"Order: {item_id} x{quantity} = ${total:.2f}")
    return {"item": item_id, "total": total}

@app.task
def apply_shipping(order_info, shipping_method="standard"):
    shipping_costs = {"standard": 5, "express": 15, "overnight": 30}
    cost = shipping_costs.get(shipping_method, 5)
    order_info["shipping"] = cost
    order_info["grand_total"] = order_info["total"] + cost
    print(f"Applied {shipping_method} shipping: ${cost}")
    return order_info

partial_order = process_order.s(item_id="ABC-123", quantity=5)
print(f"Partial signature: {partial_order}")

workflow = chain(
    process_order.s("ABC-123", 5),
    apply_shipping.s(shipping_method="express")
)

result = workflow()
print(f"Chain result: ${result['grand_total']}")

Expected output:

Partial signature: canvas.process_order(item_id='ABC-123', quantity=5)
Order: ABC-123 x5 = $500.00
Applied express shipping: $15
Chain result: $515.0

Immutable Subtasks

Prevent argument passing from previous tasks:

from celery import Celery, chain

app = Celery('signatures', broker='redis://localhost:6379/0')

@app.task
def fetch_config():
    config = {"theme": "dark", "language": "en"}
    print(f"Fetched config: {config}")
    return config

@app.task
def apply_config(config, user_id):
    result = f"Applied config for user {user_id}: {config}"
    print(result)
    return result

workflow_with_mut = chain(
    fetch_config.s(),
    apply_config.s(user_id=42)
)

print("With mutable signature (chain passes result):")
result = workflow_with_mut()
print(result)

print("\n---")

workflow_with_immut = chain(
    fetch_config.s(),
    apply_config.si({"theme": "default", "language": "es"}, user_id=99)
)

print("With immutable signature (ignores chain input):")
result = workflow_with_immut()
print(result)

Expected output:

With mutable signature (chain passes result):
Fetched config: {'theme': 'dark', 'language': 'en'}
Applied config for user 42: {'theme': 'dark', 'language': 'en'}
Applied config for user 42: {'theme': 'dark', 'language': 'en'}

---

With immutable signature (ignores chain input):
Fetched config: {'theme': 'dark', 'language': 'en'}
Applied config for user 99: {'theme': 'default', 'language': 'es'}
Applied config for user 99: {'theme': 'default', 'language': 'es'}

Common Mistakes

• Confusing .s() with .si() — .s() creates a mutable signature that receives arguments from the previous chain step. .si() creates an immutable signature that ignores chain input. Use .si() when the task doesn't need the previous result.
• Passing mutable objects in signature args — task signatures serialize arguments. Passing a mutable object (like a list or dict) that changes after the signature is created can cause unexpected behavior.
• Using partial arguments incorrectly in chains — in a chain task1.s(a) | task2.s(b), task2 receives (previous_result, b). Ensure your task signature matches this pattern.
• Not handling argument conflicts — if both the signature and the chain provide the same keyword argument, the signature's value is used. Plan argument sources carefully.
• Creating signatures inside loops with captured variables — Python closures capture variables by reference. Create signatures with current values using task.s(value) inside the loop, not task.s(i) where i changes.

Practice Questions

1. What is the difference between .s() and .si() task signatures?
2. How does argument merging work in Celery chain execution?
3. When would you use a partial task signature?
4. How do you create a task signature with keyword arguments?
5. What is the purpose of immutable subtasks?

Challenge

Build a dynamic pricing engine using task signatures. Create signature templates for: base price calculation, discount application, tax calculation, and shipping cost. Store the signatures in a database and compose them at checkout time based on the product, customer tier, and shipping method. Each step is optional — the workflow is built dynamically from available signatures.

FAQ

Mini Project

Build a workflow Builder API that accepts a list of task names and arguments from a web request and dynamically constructs a Celery canvas workflow. The API should: (1) look up each task name in a registry, (2) create signatures with provided arguments, (3) combine them into a chain or group based on a workflow type parameter, (4) execute the workflow, and (5) return the result. Support partial arguments and immutable subtasks.

What's Next

Continue with Complex Workflow Patterns to learn advanced workflow Orchestration patterns. Then explore Celery Rate Limiting for controlling task execution rates.