Celery Error Handling Patterns
In this tutorial, you will learn about Celery Error Handling Patterns. We cover key concepts, practical examples, and best practices to help you master this topic.
Implement robust error handling in Celery with retries, circuit breakers, dead letter queues, idempotency, and graceful degradation patterns.
What You Learn
You will learn production error handling patterns: retry with exponential backoff, circuit breakers for external services, dead letter queues for failed tasks, idempotent task design, and graceful degradation.
Why It Matters
Tasks fail. Networks drop. Databases timeout. Services crash. Without proper error handling, a single failure can cascade into a system-wide outage. These patterns ensure your Celery system degrades gracefully and recovers automatically.
Real-World Use
Durga Antivirus Pro uses all these patterns. External API calls use circuit breakers. Database operations retry with backoff. Malformed data tasks go to a dead letter queue for manual review. Every task is idempotent.
Circuit Breaker Pattern
from celery import Celery
import time
import functools
app = Celery('circuit', broker='redis://localhost:6379/0')
class CircuitBreaker:
def __init__(self, name, failure_threshold=5, reset_timeout=30):
self.name = name
self.failure_threshold = failure_threshold
self.reset_timeout = reset_timeout
self.failures = 0
self.last_failure = 0
self.state = 'CLOSED'
def call(self, func, *args, **kwargs):
now = time.time()
if self.state == 'OPEN':
if now - self.last_failure > self.reset_timeout:
print(f"[{self.name}] Half-open, testing...")
self.state = 'HALF_OPEN'
else:
raise Exception(f"Circuit breaker {self.name} is OPEN")
try:
result = func(*args, **kwargs)
if self.state == 'HALF_OPEN':
print(f"[{self.name}] Half-open call succeeded, closing")
self.state = 'CLOSED'
self.failures = 0
return result
except Exception as e:
self.failures += 1
self.last_failure = now
if self.failures >= self.failure_threshold:
self.state = 'OPEN'
print(f"[{self.name}] Circuit OPEN after {self.failures} failures")
raise e
api_cb = CircuitBreaker('external_api', failure_threshold=3, reset_timeout=60)
@app.task(bind=True, max_retries=5)
def call_external_api(self, url):
import requests
try:
response = api_cb.call(
lambda: requests.get(url, timeout=5)
)
return response.json()
except Exception as exc:
if isinstance(exc, Exception) and 'OPEN' in str(exc):
print(f"Circuit open, retrying in 60s")
raise self.retry(exc=exc, countdown=60)
raise self.retry(exc=exc, countdown=10)
Dead Letter Queue Pattern
from celery import Celery
import json
from datetime import datetime
app = Celery('dead_letter', broker='redis://localhost:6379/0')
DEAD_LETTER_STORE = []
def send_to_dead_letter(task, exception, context):
record = {
'task_name': task.name,
'task_id': task.request.id,
'args': task.request.args,
'kwargs': task.request.kwargs,
'error': str(exception),
'retries': task.request.retries,
'timestamp': datetime.utcnow().isoformat(),
}
DEAD_LETTER_STORE.append(record)
print(f"DEAD LETTER: {json.dumps(record, indent=2)}")
return record
class DeadLetterTask(app.Task):
"""Base task with dead letter support."""
abstract = True
max_retries = 3
autoretry_for = (Exception,)
retry_backoff = True
retry_backoff_max = 300
def on_failure(self, exc, task_id, args, kwargs, einfo):
send_to_dead_letter(self, exc, {
'task_id': task_id,
'args': args,
'kwargs': kwargs,
})
super().on_failure(exc, task_id, args, kwargs, einfo)
@app.task(base=DeadLetterTask)
def critical_operation(data):
if not data:
raise ValueError("Empty data")
return f"Processed: {data}"
# Test
critical_operation.delay("good")
critical_operation.delay(None)
Expected output:
DEAD LETTER: {
"task_name": "tasks.critical_operation",
"task_id": "550e8400-...",
"args": [null],
"kwargs": {},
"error": "Empty data",
"retries": 3,
"timestamp": "2026-06-28T10:00:00"
}
Idempotent Task Pattern
from celery import Celery
import redis
app = Celery('idempotent', broker='redis://localhost:6379/0')
dedup_client = redis.Redis.from_url('redis://localhost:6379/1')
class IdempotentTask(app.Task):
"""Base task that prevents duplicate execution."""
abstract = True
dedup_ttl = 3600 # Remember processed tasks for 1 hour
def __call__(self, *args, **kwargs):
task_id = self.request.id
# Check if already processed
dedup_key = f"dedup:{self.name}:{task_id}"
if dedup_client.get(dedup_key):
cached = dedup_client.get(f"result:{task_id}")
if cached:
return cached.decode()
return None
result = super().__call__(*args, **kwargs)
dedup_client.setex(dedup_key, self.dedup_ttl, '1')
dedup_client.setex(f"result:{task_id}", self.dedup_ttl, str(result))
return result
@app.task(base=IdempotentTask)
def process_payment(order_id, amount):
print(f"Processing payment for order {order_id}: ${amount}")
return f"Payment for order {order_id} processed"
# Even if called multiple times with same task_id, executes once
r1 = process_payment.apply_async(args=(1, 100), task_id='unique-payment-1')
try:
r1.get(timeout=10)
except Exception as e:
print(f"Expected duplicate: {e}")
Graceful Degradation Pattern
from celery import Celery
import json
app = Celery('degrade', broker='redis://localhost:6379/0')
class ServiceUnavailable(Exception):
pass
@app.task(bind=True, max_retries=2)
def generate_report(self, report_id):
"""Generate report with fallback on failure."""
def try_primary():
raise ServiceUnavailable("Primary DB is down")
def try_cache():
return "Cached report data"
def try_degraded():
return "Minimal report (degraded mode)"
strategies = [
('primary', try_primary),
('cache', try_cache),
('degraded', try_degraded),
]
for name, strategy in strategies:
try:
result = strategy()
print(f"Report {report_id}: {name} succeeded")
return {'report_id': report_id, 'source': name, 'data': result}
except ServiceUnavailable:
print(f"Report {report_id}: {name} unavailable")
continue
except Exception as e:
print(f"Report {report_id}: {name} error: {e}")
continue
raise Exception(f"Report {report_id}: all strategies failed")
Rate Limiting Pattern
from celery import Celery
import time
app = Celery('rate_limiting', broker='redis://localhost:6379/0')
import redis
rate_client = redis.Redis.from_url('redis://localhost:6379/1')
class RateLimitedTask(app.Task):
"""Base task with per-task rate limiting."""
abstract = True
max_per_minute = 60
def __call__(self, *args, **kwargs):
key = f"ratelimit:{self.name}:{int(time.time()/60)}"
count = rate_client.incr(key)
rate_client.expire(key, 120)
if count > self.max_per_minute:
raise Exception(f"Rate limit exceeded for {self.name}")
return super().__call__(*args, **kwargs)
@app.task(base=RateLimitedTask, max_per_minute=10)
def rate_limited_api(data):
return f"API call: {data}"
Common Mistakes
1. Not Making Tasks Idempotent
Celery may execute the same task multiple times (due to retries or redelivery). Without idempotency, duplicate payments, emails, or database inserts occur.
2. Retrying Permanent Errors
Some errors never recover: validation errors, permission denied, malformed data. Check exception type before retrying.
3. Not Setting Retry Limits
Without max_retries, tasks retry forever (default is 3). Set explicit limits and route permanently failed tasks to a dead letter queue.
4. Ignoring Circuit Breaker in External Calls
Calling an external API that is down causes all workers to block on timeouts. A circuit breaker fails fast and gives the service time to recover.
5. Catching All Exceptions Silently
Catching except Exception and logging without re-raising hides failures. Always re-raise or explicitly handle each exception type.
Practice Questions
1. What is the circuit breaker pattern?
A pattern that tracks failures to an external service. After N consecutive failures, the circuit opens and all calls fail immediately without attempting the operation. After a timeout, it tests one call to see if the service recovered.
2. Why should tasks be idempotent?
Because Celery guarantees at-least-once delivery. The same task may execute multiple times. Idempotency ensures duplicate executions produce the same result.
3. What is a dead letter queue?
A queue for tasks that failed permanently. Instead of retrying forever, tasks that exhaust retries are moved to a dead letter queue for manual inspection or later reprocessing.
4. How does graceful degradation work?
Define multiple strategies in priority order. If the primary Strategy fails, try the next one. The fallback may return simpler/less complete data but keeps the system running.
Challenge
Design an error handling strategy for a payment processing system. Tasks: validate card, charge card, send receipt, update ledger. Handle: card declined (no retry), network timeout (retry 3x with backoff, then dead letter), bank API down (circuit breaker, retry in 5 min), duplicate charge prevention (idempotency key).
FAQ
Mini Project: Error-Handled Task System
# error_handled_tasks.py
from celery import Celery
import json
import time
import random
app = Celery('error_handled', broker='redis://localhost:6379/0')
DEAD_LETTER = []
class RobustTask(app.Task):
abstract = True
autoretry_for = (ConnectionError, TimeoutError)
max_retries = 3
retry_backoff = True
retry_backoff_max = 60
def on_failure(self, exc, task_id, args, kwargs, einfo):
DEAD_LETTER.append({
'task': self.name,
'task_id': task_id,
'error': str(exc),
'args': args,
'kwargs': kwargs,
})
print(f"[DEAD] {self.name}[{task_id[:8]}]: {exc}")
@app.task(base=RobustTask)
def process_external_data(self, data_id):
"""Process data with full error handling."""
attempt = self.request.retries + 1
print(f"Processing {data_id} (attempt {attempt})")
if data_id == 'timeout':
raise TimeoutError("External service timed out")
if data_id == 'invalid':
raise ValueError("Invalid data format")
if data_id == 'flaky' and attempt < 3:
raise ConnectionError("Transient failure")
time.sleep(0.5)
return f"Data {data_id} processed"
@app.task(base=RobustTask)
def generate_report(self, report_type):
"""Report generation that never fails silently."""
try:
return self.process(report_type)
except ValueError:
raise
except Exception as exc:
raise self.retry(exc=exc)
def process(self, report_type):
if report_type == 'failing':
raise ValueError("Cannot generate this report")
return f"Report: {report_type}"
What's Next
Now that you understand error handling patterns, explore Celery with Django for web application integration, then learn about Celery performance tuning for optimizing throughput.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro