Celery Task Result Backend — Complete Guide
In this tutorial, you will learn about Celery Task Result Backend. We cover key concepts, practical examples, and best practices to help you master this topic.
Configure the Celery result backend to store and retrieve task return values, track task status, and handle timeouts and errors in distributed systems.
What You Learn
You will learn how to configure Redis, database, and S3 result backends, retrieve task results, track task states, handle timeouts, and manage result expiration.
Why It Matters
Without a result backend, Celery has no way to store task return values. You cannot check if a task succeeded, get its output, or debug failures. The result backend makes Celery accountable for every task it executes.
Real-World Use
Doda Browser's malware analysis stores results in a Redis backend. The web application polls task statuses and displays results when ready. Results expire after 1 hour to free Redis memory.
Configuring Result Backends
from celery import Celery
# Redis result backend
app = Celery('redis_backend',
broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
# Database backend
app2 = Celery('db_backend',
broker='redis://localhost:6379/0',
backend='db+sqlite:///results.db')
# S3 backend (needs celery[s3])
app3 = Celery('s3_backend',
broker='redis://localhost:6379/0',
backend='s3://bucket-name/results/')
# RPC backend (transient, for dev)
app4 = Celery('rpc_backend',
broker='redis://localhost:6379/0',
backend='rpc://')
# Cache backend
app5 = Celery('cache_backend',
broker='redis://localhost:6379/0',
backend='cache+memcached://127.0.0.1:11211/')
Result Backend Settings
from celery import Celery
app = Celery('result_demo', broker='redis://localhost:6379/0')
app.conf.update(
result_backend='redis://localhost:6379/1',
result_serializer='json',
result_compression='gzip',
result_expires=3600, # Results expire after 1 hour
result_extended=True, # Store traceback, args, kwargs
result_persistent=True, # Survive broker restart
task_track_started=True, # Track 'started' state
task_ignore_result=False, # Store results (default)
)
@app.task
def compute(data):
return data * 2
Retrieving Results
from celery import Celery
app = Celery('retrieve', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task(bind=True, track_started=True)
def long_task(self, seconds):
import time
for i in range(seconds):
time.sleep(1)
self.update_state(
state='PROGRESS',
meta={'current': i + 1, 'total': seconds}
)
return "Task complete"
# client.py
result = long_task.delay(10)
# Check state
print(f"State: {result.state}") # PENDING
print(f"Task ID: {result.id}")
print(f"Ready: {result.ready()}") # False
# Wait for result
output = result.get(timeout=15)
print(f"Result: {output}")
# After completion
print(f"State: {result.state}") # SUCCESS
print(f"Ready: {result.ready()}") # True
print(f"Success: {result.successful()}") # True
Expected output:
State: PENDING
Task ID: 550e8400-...
Ready: False
Result: Task complete
State: SUCCESS
Ready: True
Success: True
Error Handling with Results
from celery import Celery
app = Celery('errors', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task
def failing_task():
raise ValueError("Something went wrong")
result = failing_task.delay()
# Check if failed
print(f"Ready: {result.ready()}") # True
print(f"Failed: {result.failed()}") # True
# Get error (without raising)
print(f"Result: {result.result}") # The exception
print(f"Traceback: {result.traceback}") # The traceback string
# Get with propagate=False
try:
data = result.get(propagate=False)
print(f"Got result: {data}")
except Exception as e:
print(f"Exception: {e}")
Expected output:
Ready: True
Failed: True
Result: ValueError('Something went wrong')
Traceback: Traceback (most recent call last)...
Got result: ValueError('Something went wrong')
AsyncResult and Result Sets
from celery import Celery
from celery.result import AsyncResult, ResultSet
app = Celery('async', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
@app.task
def double(n):
return n * 2
# Single result
result = double.delay(21)
async_result = AsyncResult(result.id, app=app)
print(f"Status: {async_result.status}")
print(f"Result: {async_result.get(timeout=5)}")
# Multiple results
results = [double.delay(i) for i in range(5)]
result_set = ResultSet(results)
print(f"Waiting for {len(result_set)} tasks...")
print(f"Results: {result_set.get(timeout=10)}")
print(f"All done: {result_set.completed_count()}")
Expected output:
Status: SUCCESS
Result: 42
Waiting for 5 tasks...
Results: [0, 2, 4, 6, 8]
All done: 5
Result Expiration and Cleanup
from celery import Celery
import redis
app = Celery('cleanup', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
# Results expire after configured time
app.conf.result_expires = 3600 # 1 hour
@app.task
def temp_result(data):
return f"Temporary: {data}"
# Manual cleanup
def cleanup_old_results():
client = redis.Redis.from_url(app.conf.result_backend)
count = 0
for key in client.scan_iter("celery-task-meta-*"):
# Check task age from stored data
count += 1
print(f"Found {count} stored results")
return count
# Delete specific result
def delete_result(task_id):
client = redis.Redis.from_url(app.conf.result_backend)
client.delete(f"celery-task-meta-{task_id}")
print(f"Deleted result for {task_id}")
Common Mistakes
1. Forgetting to Set result_backend
Without a result backend, result.get() returns None or raises NotImplementedError. Always configure the backend if you need results.
2. Not Setting result_expires
Results accumulate in the backend forever. Without expiration, Redis memory fills up. Set result_expires=3600 for auto-cleanup.
3. Using Same Redis Database for Broker and Backend
Using the same database (e.g., both db0) causes key collisions between task messages and results. Use different databases.
4. Blocking on .get() Without Timeout
.get() without timeout blocks indefinitely if the task hangs. Always set timeout to prevent worker hangs.
5. Storing Large Results
Large result objects consume backend memory. For large outputs (images, files), store them externally and return a reference (URL or path).
Practice Questions
1. What backends does Celery support for results?
Redis, database (SQLite, PostgreSQL, MySQL), S3, RPC, cache (Memcached), and custom backends.
2. What does result_expires do?
Automatically deletes task results after the specified time. Prevents the result backend from accumulating stale data indefinitely.
3. How do you retrieve a task result without raising an exception?
Use result.get(propagate=False). It returns the exception object instead of raising it. Check result.failed() to see if the task failed.
4. What is task_track_started used for?
When enabled, the task state changes to 'STARTED' when execution begins. Without it, the state jumps from PENDING directly to SUCCESS or FAILURE.
Challenge
Design a result backend Strategy for a document processing system. Tasks: PDF generation (large output, store in S3, return URL), text analysis (small output, store in Redis with 24h expiry), error tracking (persist in database indefinitely for auditing), and progress tracking (ephemeral, update via Websocket).
FAQ
Mini Project: Result Backend Demo
# result_demo.py
from celery import Celery
from celery.result import AsyncResult
import time
import json
app = Celery('result_system', broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
app.conf.update(
result_serializer='json',
result_expires=3600,
task_track_started=True,
task_serializer='json',
accept_content=['json'],
)
@app.task(bind=True)
def process_task(self, task_type, data):
self.update_state(state='PROCESSING', meta={'type': task_type})
time.sleep(2)
if task_type == 'fail':
raise RuntimeError(f"Failed to process {data}")
return {
'task_type': task_type,
'input': data,
'output': data.upper(),
'processed_at': '2026-06-28T10:00:00Z'
}
def get_status(task_id):
result = AsyncResult(task_id, app=app)
info = {
'task_id': task_id,
'status': result.status,
'ready': result.ready(),
'successful': result.successful(),
'failed': result.failed(),
}
if result.ready():
info['result'] = result.result
info['traceback'] = result.traceback
return info
if __name__ == '__main__':
r1 = process_task.delay('normal', 'hello')
r2 = process_task.delay('fail', 'crash_me')
time.sleep(1)
print("Before completion:")
print(json.dumps(get_status(r1.id), indent=2))
print()
time.sleep(3)
print("After completion:")
print(json.dumps(get_status(r1.id), indent=2))
print(json.dumps(get_status(r2.id), indent=2))
Expected output:
Before completion:
{
"task_id": "550e8400-...",
"status": "PROCESSING",
"ready": false,
"successful": false,
"failed": false
}
After completion:
{
"task_id": "550e8400-...",
"status": "SUCCESS",
"ready": true,
"successful": true,
"failed": false,
"result": {"task_type": "normal", "input": "hello", "output": "HELLO", ...}
}
What's Next
Now that you understand result backends, explore task chaining for building multi-step workflows, then learn about task groups for parallel task execution.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro