Task Chaining in Celery — Complete Guide
In this tutorial, you will learn about Task Chaining in Celery. We cover key concepts, practical examples, and best practices to help you master this topic.
Chain Celery tasks to create sequential workflows where the output of one task becomes the input of the next for complex processing pipelines.
What You Learn
You will learn how to create task chains, use the pipe operator, combine chains with groups and chords, handle errors in chains, and build complex processing pipelines.
Why It Matters
Real-world applications rarely consist of single tasks. Processing an image requires resizing, analyzing, and generating thumbnails. Chains let you compose these steps into reliable, testable workflows without manually managing intermediate results.
Real-World Use
Doda Browser's file analysis pipeline uses a chain: validate_file | scan_for_malware | generate_report | notify_user. Each step depends on the previous one. If any step fails, the chain stops and the error is logged.
Basic Chain
from celery import Celery
app = Celery('chain_demo', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task
def add(x, y):
return x + y
@app.task
def multiply(result, factor):
return result * factor
@app.task
def format_result(result):
return f"Final result: {result}"
# Chain: add(2, 2) -> multiply(*, 10) -> format_result(*)
from celery import chain
result = chain(
add.s(2, 2),
multiply.s(10),
format_result.s()
)()
print(f"Chain result: {result.get(timeout=10)}")
Expected output:
Chain result: Final result: 40
Pipe Operator
from celery import Celery
app = Celery('pipe_demo', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task
def validate(data):
if not data:
raise ValueError("Empty data")
return data.upper()
@app.task
def process(data):
return f"Processed: {data}"
@app.task
def save(data):
return f"Saved: {data}"
# Using pipe operator
result = (validate.s("test_data") | process.s() | save.s())()
print(f"Pipe result: {result.get(timeout=10)}")
# Chain with arguments at different points
result2 = (add.s(5, 3) | multiply.s(4))()
print(f"Calc result: {result2.get(timeout=10)}")
# Import add from above
Expected output:
Pipe result: Saved: Processed: TEST_DATA
Calc result: 32
Partial Chains
Supply some arguments upfront and let the chain fill in the rest:
from celery import Celery
app = Celery('partial', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task
def scale_image(image_path, width, height):
return f"scaled/{image_path}"
@app.task
def add_watermark(image_path, text):
return f"watermarked/{image_path}"
@app.task
def upload(image_path, bucket):
return f"https://{bucket}/{image_path}"
# Partial chain - bucket known upfront, image_path comes from previous step
process_image = (
scale_image.s(width=800, height=600) |
add_watermark.s(text="DodaTech") |
upload.s(bucket="dodatech-images")
)
# Execute with the starting argument
result = process_image.delay("photo.jpg")
print(result.get(timeout=10))
Expected output:
https://dodatech-images/watermarked/scaled/photo.jpg
Chain with Error Handling
from celery import Celery
from celery import chain
app = Celery('chain_errors', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task(bind=True)
def step1(self, data):
print(f"Step 1: {data}")
if data == "fail":
raise ValueError("Step 1 failed")
return f"step1_{data}"
@app.task(bind=True)
def step2(self, data):
print(f"Step 2: {data}")
return f"step2_{data}"
@app.task(bind=True)
def step3(self, data):
print(f"Step 3: {data}")
return f"step3_{data}"
@app.task
def error_handler(request, exc, traceback):
print(f"Chain failed at {request.task}: {exc}")
return {"error": str(exc), "task": request.task}
# Working chain
result = chain(step1.s("good"), step2.s(), step3.s())()
print(f"Success: {result.get(timeout=10)}")
# Failing chain - stops at step1
result2 = chain(
step1.s("fail"),
step2.s(),
step3.s()
)()
try:
result2.get(timeout=10)
except ValueError as e:
print(f"Chain error caught: {e}")
Expected output:
Step 1: good
Step 2: step1_good
Step 3: step2_step1_good
Success: step3_step2_step1_good
Step 1: fail
Chain error caught: Step 1 failed
Chain with Callbacks
from celery import Celery
from celery import chain, group
app = Celery('callbacks', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task
def fetch_data(query):
return f"data_for_{query}"
@app.task
def process_data(data):
return f"processed_{data}"
@app.task
def save_result(data):
return f"saved_{data}"
@app.task
def notify(data):
return f"notification_sent"
# Chain with success callback
main_chain = fetch_data.s("users") | process_data.s() | save_result.s()
# Callback executes after main chain completes
full_workflow = main_chain | notify.s()
result = full_workflow.delay()
print(f"Final: {result.get(timeout=10)}")
Expected output:
Final: notification_sent
Immutable Signatures
Use .si() for immutable signatures where arguments from the previous task are ignored:
from celery import Celery
app = Celery('immutable', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task
def fetch_page(url):
return f"content_of_{url}"
@app.task
def analyze_text(text):
return f"analysis_of_{len(text)}_chars"
@app.task
def log_metrics():
"""This task ignores its input and always runs the same way."""
return "metrics_logged"
# log_metrics receives the output of analyze_text but ignores it
# Use .si() to make it immutable (doesn't accept previous result)
pipeline = fetch_page.s("https://example.com") | analyze_text.s() | log_metrics.si()
result = pipeline.delay()
print(result.get(timeout=10))
Expected output:
metrics_logged
Complex Chain Example
from celery import Celery, chain, group, chord
import time
app = Celery('complex', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task
def fetch_user(user_id):
return {'id': user_id, 'name': f'User_{user_id}', 'items': [1, 2, 3]}
@app.task
def process_items(user_data):
items = user_data['items']
return [f"processed_{item}" for item in items]
@app.task
def save_each(item):
time.sleep(0.5)
return f"saved_{item}"
@app.task
def combine(results):
return {"total": len(results), "items": results}
@app.task
def notify_user(user_id):
return f"User {user_id} notified"
# Complex workflow
def user_workflow(user_id):
workflow = (
fetch_user.s(user_id) |
process_items.s() |
group([save_each.s(item) for item in range(3)]) |
combine.s() |
notify_user.s(user_id)
)
return workflow.delay()
result = user_workflow(42)
print(result.get(timeout=30))
Common Mistakes
1. Forgetting That Chains Pass Results
Each task in a chain receives the previous task's return value as the first argument. Design tasks to accept and use this.
2. Using .s() vs .si() Incorrectly
.s() creates a signature that receives the previous result. .si() creates an immutable signature that ignores it. Use .si() when the task does not need the previous output.
3. Chains with Variable-Length Groups
When a task returns a list and you want to Process each item in parallel, combine the list-returning task with a group. Use chord for the final aggregation.
4. Not Handling Chain Failures
A failure in any task stops the entire chain. Use link_error callbacks or catch exceptions at the end of the chain to handle errors gracefully.
5. Creating Overly Long Chains
Chains longer than 5-10 tasks are hard to debug and test. Break them into sub-chains with intermediate result storage.
Practice Questions
1. What is a Celery chain?
A sequence of tasks where the output of each task is passed as input to the next task. Created with chain() or the | pipe operator.
2. What is the difference between .s() and .si()?
.s() creates a signature that receives and passes forward the previous result. .si() creates an immutable signature that does not accept the previous result.
3. How do you handle errors in a chain?
Use link_error callback on the chain or catch exceptions in the final task. Chain stops on first failure by default.
4. What happens if a task in the middle of a chain fails?
The chain stops. Subsequent tasks are not executed. The error propagates to the chain's result.
Challenge
Build a document processing pipeline: validate_format | extract_text | analyze_sentiment | generate_summary | save_to_database | notify_user. Each task can fail. Implement error handling that logs failures, sends alerts on critical errors, and retries transient failures within the chain.
FAQ
Mini Project: Image Processing Pipeline
# image_pipeline.py
from celery import Celery, chain, group
import time
app = Celery('image_pipe', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task(bind=True)
def download_image(self, url):
print(f"Downloading: {url}")
time.sleep(1)
return f"/tmp/{url.split('/')[-1]}"
@app.task(bind=True)
def validate_image(self, path):
print(f"Validating: {path}")
time.sleep(0.5)
return path
@app.task(bind=True)
def resize_image(self, path, width=800, height=600):
print(f"Resizing {path} to {width}x{height}")
time.sleep(1)
return f"{path}_resized_{width}x{height}"
@app.task(bind=True)
def generate_thumbnail(self, path, size=150):
print(f"Thumbnail {path} at {size}x{size}")
time.sleep(0.5)
return f"{path}_thumb_{size}"
@app.task(bind=True)
def upload_image(self, path):
print(f"Uploading: {path}")
time.sleep(0.5)
return f"https://cdn.dodatech.com/{path.split('/')[-1]}"
@app.task(bind=True)
def cleanup_temp(self, path):
print(f"Cleaning up: {path}")
time.sleep(0.2)
return "cleaned"
# Image processing pipeline
def process_image(url):
pipeline = (
download_image.s(url) |
validate_image.s() |
group(
resize_image.s(800, 600),
generate_thumbnail.s(150),
) |
group(
upload_image.s(),
upload_image.s(),
)
)
return pipeline.delay()
# Execute
result = process_image("https://example.com/photo.jpg")
print(f"Pipeline started: {result.id}")
# Note: this complex pipeline with groups inside chains requires careful setup
# Simpler linear chain works reliably
What's Next
Now that you understand task chaining, explore task groups and chords for parallel task execution, then learn about monitoring with Flower for tracking task execution.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro