Table of Contents
Quick Navigation
Celery Architecture
System Components
┌─────────────────────────────────────────────────────────┐
│ Celery Architecture │
└─────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────┐
│ Application/Producer │
│ ┌──────────────────────────────────────────────────┐ │
│ │ @app.task │ │
│ │ def process_data(data): │ │
│ │ # Task code │ │
│ │ │ │
│ │ # Send task to queue │ │
│ │ process_data.delay(data) │ │
│ └──────────────────────────────────────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
│ Publishes task
▼
┌─────────────────────────────────────────────────────────┐
│ Message Broker (Queue) │
│ ┌──────────────────────────────────────────────────┐ │
│ │ Redis / RabbitMQ / Amazon SQS │ │
│ │ │ │
│ │ Queues: │ │
│ │ • default │ │
│ │ • high_priority │ │
│ │ • email │ │
│ │ • processing │ │
│ └──────────────────────────────────────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
│ Workers consume tasks
▼
┌─────────────────────────────────────────────────────────┐
│ Celery Workers │
│ ┌────────────────────┐ ┌────────────────────┐ │
│ │ Worker Pool 1 │ │ Worker Pool 2 │ │
│ │ ┌──────────────┐ │ │ ┌──────────────┐ │ │
│ │ │ Process 1 │ │ │ │ Process 1 │ │ │
│ │ │ Process 2 │ │ │ │ Process 2 │ │ │
│ │ │ Process 3 │ │ │ │ Process 3 │ │ │
│ │ │ Process 4 │ │ │ │ Process 4 │ │ │
│ │ └──────────────┘ │ │ └──────────────┘ │ │
│ └────────────────────┘ └────────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
│ Stores results
▼
┌─────────────────────────────────────────────────────────┐
│ Result Backend │
│ ┌──────────────────────────────────────────────────┐ │
│ │ Redis / Database / Memcached │ │
│ │ │ │
│ │ Stores: │ │
│ │ • Task results │ │
│ │ • Task status │ │
│ │ • Task metadata │ │
│ └──────────────────────────────────────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Beat Scheduler (Optional) │
│ ┌──────────────────────────────────────────────────┐ │
│ │ Celery Beat │ │
│ │ │ │
│ │ Schedules periodic tasks: │ │
│ │ • Every 5 minutes │ │
│ │ • Daily at midnight │ │
│ │ • Every Monday at 9am │ │
│ └──────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘
Key Concepts:
1. Task: Unit of work to be executed asynchronously
2. Queue: Named channel where tasks wait for execution
3. Worker: Process that executes tasks from queues
4. Broker: Message queue system (Redis/RabbitMQ)
5. Result Backend: Stores task results and state
6. Beat: Scheduler for periodic tasks
7. Producer: Code that sends tasks to queues
8. Consumer: Worker that receives and executes tasks
Task Lifecycle
┌─────────────────────────────────────────────────────────┐
│ Task Lifecycle │
└─────────────────────────────────────────────────────────┘
1. PENDING
↓
Task is created and sent to broker
↓
2. RECEIVED (optional)
↓
Worker acknowledges receipt
↓
3. STARTED
↓
Worker begins executing task
↓
4. Processing...
↓
┌──────────────┬──────────────┬──────────────┐
│ │ │ │
▼ ▼ ▼ ▼
SUCCESS FAILURE RETRY REVOKED
(completed) (error) (try again) (cancelled)
State Transitions:
PENDING → Task waiting in queue
RECEIVED → Worker has received task
STARTED → Worker executing task
RETRY → Task failed, will retry
FAILURE → Task failed permanently
SUCCESS → Task completed successfully
REVOKED → Task cancelled before execution
Task States in Code:
from celery import states
# Check task state
if task.state == states.PENDING:
print("Task is waiting")
elif task.state == states.STARTED:
print("Task is running")
elif task.state == states.SUCCESS:
print("Task completed")
result = task.result
elif task.state == states.FAILURE:
print("Task failed")
error = task.result # Exception object
Custom States:
@app.task(bind=True)
def long_task(self, items):
total = len(items)
for i, item in enumerate(items):
# Update custom state
self.update_state(
state='PROGRESS',
meta={
'current': i + 1,
'total': total,
'percent': int((i + 1) / total * 100)
}
)
process_item(item)
return {'status': 'complete', 'total': total}
Setup & Configuration
Installation & Basic Setup
# Installation
pip install celery
# With Redis
pip install celery[redis]
# With RabbitMQ
pip install celery[amqp]
# With all extras
pip install celery[redis,auth,msgpack]
# Basic Celery app (celery_app.py)
from celery import Celery
app = Celery(
'myapp',
broker='redis://localhost:6379/0',
backend='redis://localhost:6379/0'
)
# Basic task
@app.task
def add(x, y):
return x + y
# Run worker
# celery -A celery_app worker --loglevel=info
# Configuration file (celeryconfig.py)
# Broker settings
broker_url = 'redis://localhost:6379/0'
broker_connection_retry_on_startup = True
# Result backend
result_backend = 'redis://localhost:6379/0'
result_expires = 3600 # 1 hour
# Task settings
task_serializer = 'json'
accept_content = ['json']
result_serializer = 'json'
timezone = 'UTC'
enable_utc = True
# Task execution
task_track_started = True
task_time_limit = 30 * 60 # 30 minutes
task_soft_time_limit = 25 * 60 # 25 minutes
task_acks_late = True
worker_prefetch_multiplier = 1
# Worker settings
worker_max_tasks_per_child = 1000
worker_disable_rate_limits = False
# Load config
app.config_from_object('celeryconfig')
# Advanced configuration
from celery import Celery
app = Celery('myapp')
app.conf.update(
# Broker
broker_url='redis://localhost:6379/0',
broker_connection_retry=True,
broker_connection_retry_on_startup=True,
broker_connection_max_retries=10,
# Results
result_backend='redis://localhost:6379/0',
result_expires=3600,
result_extended=True,
# Serialization
task_serializer='json',
result_serializer='json',
accept_content=['json'],
# Time
timezone='America/New_York',
enable_utc=True,
# Task routing
task_routes={
'myapp.tasks.send_email': {'queue': 'email'},
'myapp.tasks.process_data': {'queue': 'processing'},
},
# Task priorities
task_queue_max_priority=10,
task_default_priority=5,
# Rate limiting
task_annotations={
'myapp.tasks.send_email': {'rate_limit': '100/h'},
'myapp.tasks.api_call': {'rate_limit': '10/s'},
},
# Monitoring
worker_send_task_events=True,
task_send_sent_event=True,
# Security
task_reject_on_worker_lost=True,
task_acks_late=True,
)
# Environment-based config
import os
class Config:
broker_url = os.getenv(
'CELERY_BROKER_URL',
'redis://localhost:6379/0'
)
result_backend = os.getenv(
'CELERY_RESULT_BACKEND',
'redis://localhost:6379/0'
)
class DevelopmentConfig(Config):
task_always_eager = True # Execute tasks synchronously
task_eager_propagates = True
class ProductionConfig(Config):
worker_max_tasks_per_child = 1000
worker_prefetch_multiplier = 4
config = {
'development': DevelopmentConfig,
'production': ProductionConfig,
}
app.config_from_object(config[os.getenv('ENV', 'development')])
Project Structure
myproject/
├── celery_app.py # Celery application
├── celeryconfig.py # Configuration
├── tasks/
│ ├── __init__.py
│ ├── email.py # Email tasks
│ ├── processing.py # Data processing tasks
│ └── maintenance.py # Cleanup tasks
├── schedules/
│ └── periodic.py # Periodic task definitions
├── utils/
│ ├── monitoring.py # Custom monitoring
│ └── helpers.py # Task helpers
└── docker-compose.yml # Services setup
# celery_app.py
from celery import Celery
app = Celery('myproject')
app.config_from_object('celeryconfig')
# Auto-discover tasks
app.autodiscover_tasks(['tasks'])
# tasks/email.py
from celery_app import app
from celery.utils.log import get_task_logger
logger = get_task_logger(__name__)
@app.task(bind=True, max_retries=3)
def send_email(self, recipient, subject, body):
try:
logger.info(f"Sending email to {recipient}")
# Email sending logic
return {'status': 'sent', 'recipient': recipient}
except Exception as exc:
logger.error(f"Failed to send email: {exc}")
raise self.retry(exc=exc, countdown=60)
# tasks/processing.py
from celery_app import app
@app.task(bind=True)
def process_data(self, data_id):
# Update state
self.update_state(
state='PROGRESS',
meta={'current': 0, 'total': 100}
)
# Process data
result = heavy_processing(data_id)
return {'status': 'completed', 'result': result}
# Starting workers
# General worker
celery -A celery_app worker --loglevel=info
# Worker for specific queue
celery -A celery_app worker -Q email --loglevel=info
# Worker with concurrency
celery -A celery_app worker --concurrency=4
# Worker with autoscaling
celery -A celery_app worker --autoscale=10,3
# Multiple workers
celery multi start w1 w2 w3 -A celery_app -l info
# Stop workers
celery multi stop w1 w2 w3
# Restart workers
celery multi restart w1 w2 w3 -A celery_app
# Beat scheduler
celery -A celery_app beat --loglevel=info
# With persistent scheduler
celery -A celery_app beat --scheduler django_celery_beat.schedulers:DatabaseScheduler
Task Patterns
Task Definition & Execution
# Basic task
from celery_app import app
@app.task
def add(x, y):
return x + y
# Call synchronously (blocks until complete)
result = add(4, 5)
# Call asynchronously
task = add.delay(4, 5)
# Call with apply_async (more options)
task = add.apply_async(args=[4, 5])
# Task with bind=True (access to task instance)
@app.task(bind=True)
def debug_task(self):
print(f'Request: {self.request!r}')
print(f'Task ID: {self.request.id}')
print(f'Task Name: {self.request.task}')
print(f'Args: {self.request.args}')
print(f'Kwargs: {self.request.kwargs}')
# Task with custom name
@app.task(name='tasks.add_numbers')
def add(x, y):
return x + y
# Task options
@app.task(
bind=True,
max_retries=3,
default_retry_delay=60,
rate_limit='100/h',
time_limit=300,
soft_time_limit=240,
ignore_result=False,
track_started=True,
acks_late=True,
reject_on_worker_lost=True,
)
def process_order(self, order_id):
# Task logic
pass
# Apply async with options
task = process_order.apply_async(
args=[order_id],
kwargs={'priority': 'high'},
queue='orders',
priority=9,
countdown=10, # Execute after 10 seconds
eta=datetime.utcnow() + timedelta(hours=1), # Execute at specific time
expires=300, # Expire if not executed in 5 minutes
retry=True,
retry_policy={
'max_retries': 3,
'interval_start': 0,
'interval_step': 0.2,
'interval_max': 0.2,
}
)
# Check task result
if task.ready():
result = task.get()
print(f"Result: {result}")
else:
print("Task not ready yet")
# Wait for result (blocks)
result = task.get(timeout=10)
# Check if task succeeded
if task.successful():
print("Task completed successfully")
# Check if task failed
if task.failed():
print("Task failed")
print(task.traceback)
# Task with progress updates
@app.task(bind=True)
def process_items(self, items):
total = len(items)
for i, item in enumerate(items):
# Update progress
self.update_state(
state='PROGRESS',
meta={
'current': i + 1,
'total': total,
'percent': int((i + 1) / total * 100),
'status': f'Processing item {i + 1} of {total}'
}
)
time.sleep(1) # Simulate work
return {'status': 'Complete', 'total': total}
# Monitor progress
task = process_items.delay(list(range(100)))
while not task.ready():
if task.state == 'PROGRESS':
meta = task.info
print(f"{meta['percent']}% - {meta['status']}")
time.sleep(1)
# Task with callback
@app.task
def on_success_callback(result):
print(f"Task succeeded with result: {result}")
@app.task
def on_failure_callback(exc, task_id, args, kwargs, einfo):
print(f"Task {task_id} failed: {exc}")
task = process_order.apply_async(
args=[123],
link=on_success_callback.s(),
link_error=on_failure_callback.s()
)
# Task with context manager
from contextlib import contextmanager
@contextmanager
def task_context():
# Setup
print("Starting task context")
connection = get_db_connection()
try:
yield connection
finally:
# Cleanup
connection.close()
print("Closing task context")
@app.task
def process_with_context(data_id):
with task_context() as conn:
# Use connection
data = conn.fetch(data_id)
return process(data)
# Shared task (for reusability)
from celery import shared_task
@shared_task
def send_notification(user_id, message):
# Can be used across multiple Celery apps
pass
# Task with custom result
from celery.result import AsyncResult
@app.task
def long_running_task():
# Do work
return {'status': 'success', 'data': result_data}
# Get task result
task_id = 'task-id-here'
result = AsyncResult(task_id, app=app)
if result.ready():
data = result.get()
print(data)
# Revoke (cancel) task
# Cancel without executing
task.revoke()
# Cancel and terminate if already running
task.revoke(terminate=True)
# Cancel with signal
task.revoke(terminate=True, signal='SIGKILL')
# Task groups
from celery import group
# Execute multiple tasks in parallel
job = group(
add.s(2, 2),
add.s(4, 4),
add.s(8, 8)
)
result = job.apply_async()
# Wait for all results
results = result.get()
print(results) # [4, 8, 16]
Advanced Task Patterns
# Task inheritance
from celery import Task
class CallbackTask(Task):
def on_success(self, retval, task_id, args, kwargs):
print(f"Task {task_id} succeeded")
def on_failure(self, exc, task_id, args, kwargs, einfo):
print(f"Task {task_id} failed: {exc}")
def on_retry(self, exc, task_id, args, kwargs, einfo):
print(f"Task {task_id} retrying")
@app.task(base=CallbackTask)
def process_data(data):
# Task logic
return result
# Dynamically create tasks
def create_task(name, func):
@app.task(name=name)
def dynamic_task(*args, **kwargs):
return func(*args, **kwargs)
return dynamic_task
# Create tasks dynamically
for operation in ['add', 'multiply', 'divide']:
task = create_task(f'math.{operation}', operations[operation])
# Task with request context
@app.task(bind=True)
def contextualized_task(self):
# Access request info
task_id = self.request.id
task_name = self.request.task
args = self.request.args
kwargs = self.request.kwargs
retries = self.request.retries
eta = self.request.eta
expires = self.request.expires
return {
'task_id': task_id,
'task_name': task_name,
'retries': retries
}
# Immutable signatures
from celery import signature
# Create signature
sig = add.signature((2, 2), immutable=True)
# Or shorthand
sig = add.si(2, 2)
# Execute
result = sig.apply_async()
# Partial signatures
# Create partial
partial = add.s(2)
# Complete and execute
result = partial.apply_async(args=[2]) # add(2, 2)
# Task with database transaction
from sqlalchemy.orm import Session
@app.task(bind=True)
def transactional_task(self, data):
session = Session()
try:
# Do database work
obj = MyModel(**data)
session.add(obj)
session.commit()
return {'id': obj.id}
except Exception as exc:
session.rollback()
raise self.retry(exc=exc, countdown=60)
finally:
session.close()
# Task with locking
from redis import Redis
from redis.lock import Lock
redis_client = Redis()
@app.task(bind=True)
def exclusive_task(self, resource_id):
lock_id = f'lock:{resource_id}'
lock = Lock(redis_client, lock_id, timeout=300)
if lock.acquire(blocking=False):
try:
# Do exclusive work
return process_resource(resource_id)
finally:
lock.release()
else:
# Resource locked, retry later
raise self.retry(countdown=10)
# Task with idempotency check
@app.task(bind=True)
def idempotent_task(self, operation_id, data):
# Check if already processed
if redis_client.exists(f'processed:{operation_id}'):
return {'status': 'already_processed'}
# Process
result = process(data)
# Mark as processed
redis_client.setex(
f'processed:{operation_id}',
3600, # Expire after 1 hour
'true'
)
return result
# Task batching
from collections import defaultdict
batch_buffer = defaultdict(list)
BATCH_SIZE = 100
@app.task
def batched_task(item):
batch_id = item['batch_id']
batch_buffer[batch_id].append(item)
if len(batch_buffer[batch_id]) >= BATCH_SIZE:
items = batch_buffer.pop(batch_id)
return process_batch(items)
return {'status': 'buffered'}
# Task with file processing
import tempfile
import os
@app.task(bind=True)
def process_file(self, file_url):
# Download to temp file
with tempfile.NamedTemporaryFile(delete=False) as tmp:
download_file(file_url, tmp.name)
tmp_path = tmp.name
try:
# Process file
result = process_large_file(tmp_path)
return result
finally:
# Cleanup
os.unlink(tmp_path)
# Task with exponential backoff
@app.task(bind=True, max_retries=5)
def api_call_with_backoff(self, url, data):
try:
response = requests.post(url, json=data)
response.raise_for_status()
return response.json()
except requests.RequestException as exc:
# Exponential backoff: 2^retry seconds
countdown = 2 ** self.request.retries
raise self.retry(exc=exc, countdown=countdown)
Periodic Tasks
Beat Scheduler Configuration
# Basic periodic task
from celery import Celery
from celery.schedules import crontab
app = Celery('myapp')
app.conf.beat_schedule = {
# Execute every 30 seconds
'add-every-30-seconds': {
'task': 'tasks.add',
'schedule': 30.0,
'args': (16, 16)
},
# Execute every 5 minutes
'process-data-every-5-minutes': {
'task': 'tasks.process_data',
'schedule': crontab(minute='*/5'),
},
# Execute daily at midnight
'cleanup-daily': {
'task': 'tasks.cleanup',
'schedule': crontab(hour=0, minute=0),
},
# Execute every Monday at 8am
'weekly-report': {
'task': 'tasks.generate_report',
'schedule': crontab(hour=8, minute=0, day_of_week=1),
},
# Execute on specific days
'weekend-task': {
'task': 'tasks.weekend_job',
'schedule': crontab(hour=10, minute=0, day_of_week='sat,sun'),
},
# Execute first day of month
'monthly-billing': {
'task': 'tasks.billing',
'schedule': crontab(hour=0, minute=0, day_of_month=1),
},
}
# Crontab patterns
from celery.schedules import crontab
# Every minute
crontab()
# Every 15 minutes
crontab(minute='*/15')
# Every hour
crontab(minute=0)
# Every day at 3:30 AM
crontab(hour=3, minute=30)
# Every Monday at 9 AM
crontab(hour=9, minute=0, day_of_week=1)
# Every weekday at 5 PM
crontab(hour=17, minute=0, day_of_week='1-5')
# First day of every month
crontab(hour=0, minute=0, day_of_month=1)
# Every quarter (Jan, Apr, Jul, Oct) at 1st, midnight
crontab(hour=0, minute=0, day_of_month=1, month_of_year='1,4,7,10')
# Last day of month (approximately)
crontab(hour=0, minute=0, day_of_month='28-31')
# Solar schedules
from celery.schedules import solar
app.conf.beat_schedule = {
# Execute at sunrise
'sunrise-task': {
'task': 'tasks.morning_routine',
'schedule': solar('sunrise', -37.81, 144.96), # Melbourne
},
# Execute at sunset
'sunset-task': {
'task': 'tasks.evening_routine',
'schedule': solar('sunset', 40.71, -74.00), # New York
},
}
# Advanced scheduling
from datetime import timedelta
app.conf.beat_schedule = {
# Every 30 minutes
'every-30-minutes': {
'task': 'tasks.check_status',
'schedule': timedelta(minutes=30),
},
# Every 2 hours with arguments
'process-with-args': {
'task': 'tasks.process',
'schedule': timedelta(hours=2),
'args': ('arg1', 'arg2'),
'kwargs': {'key': 'value'},
},
# With options
'task-with-options': {
'task': 'tasks.important',
'schedule': crontab(minute=0, hour='*/3'),
'options': {
'queue': 'priority',
'priority': 10,
'expires': 300,
}
},
}
# Dynamic periodic tasks (django-celery-beat)
# Install: pip install django-celery-beat
# settings.py
INSTALLED_APPS = [
'django_celery_beat',
]
# Run migrations
python manage.py migrate django_celery_beat
# Use database scheduler
celery -A myapp beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler
# Create periodic task programmatically
from django_celery_beat.models import PeriodicTask, IntervalSchedule
import json
# Create interval (every 10 seconds)
schedule, created = IntervalSchedule.objects.get_or_create(
every=10,
period=IntervalSchedule.SECONDS,
)
# Create task
PeriodicTask.objects.create(
interval=schedule,
name='Process data every 10 seconds',
task='tasks.process_data',
args=json.dumps(['arg1', 'arg2']),
)
# Crontab schedule
from django_celery_beat.models import CrontabSchedule
schedule, created = CrontabSchedule.objects.get_or_create(
minute='0',
hour='*/4',
day_of_week='*',
day_of_month='*',
month_of_year='*',
)
PeriodicTask.objects.create(
crontab=schedule,
name='Task every 4 hours',
task='tasks.periodic_task',
)
# Disable/enable periodic tasks
task = PeriodicTask.objects.get(name='my-task')
task.enabled = False
task.save()
# One-off scheduled tasks
from datetime import datetime, timedelta
PeriodicTask.objects.create(
name='One time task',
task='tasks.one_time',
one_off=True,
start_time=datetime.utcnow() + timedelta(hours=1),
)
# Custom schedule
from celery.schedules import schedule
class CustomSchedule(schedule):
def is_due(self, last_run_at):
# Custom logic to determine if task should run
if custom_condition():
return True, 60 # (should_run, next_check_in_seconds)
return False, 60
app.conf.beat_schedule = {
'custom-schedule': {
'task': 'tasks.custom',
'schedule': CustomSchedule(),
},
}
Task Routing
Queue Configuration
# Define queues
from kombu import Queue, Exchange
app.conf.task_queues = (
Queue('default', Exchange('default'), routing_key='default'),
Queue('high_priority', Exchange('high_priority'), routing_key='high_priority'),
Queue('low_priority', Exchange('low_priority'), routing_key='low_priority'),
Queue('email', Exchange('email'), routing_key='email'),
Queue('processing', Exchange('processing'), routing_key='processing'),
)
# Default queue
app.conf.task_default_queue = 'default'
app.conf.task_default_exchange = 'default'
app.conf.task_default_routing_key = 'default'
# Route tasks to queues
app.conf.task_routes = {
# Route by task name
'tasks.send_email': {'queue': 'email'},
'tasks.send_sms': {'queue': 'email'},
# Pattern matching
'tasks.process_*': {'queue': 'processing'},
# With priority
'tasks.critical': {
'queue': 'high_priority',
'priority': 10,
},
# With rate limit
'tasks.api_call': {
'queue': 'default',
'rate_limit': '100/m',
},
}
# Route with function
def route_task(name, args, kwargs, options, task=None, **kw):
if 'important' in kwargs:
return {'queue': 'high_priority'}
return {'queue': 'default'}
app.conf.task_routes = (route_task,)
# Send task to specific queue
# Using routing_key
task = process_data.apply_async(
args=[data],
queue='processing'
)
# Multiple queues
task = process_data.apply_async(
args=[data],
queue='processing',
routing_key='processing.urgent'
)
# Priority queues
app.conf.task_queue_max_priority = 10
app.conf.task_default_priority = 5
# Send with priority
task = process_data.apply_async(
args=[data],
priority=9 # High priority
)
# Worker consuming specific queues
# Single queue
celery -A myapp worker -Q email -l info
# Multiple queues
celery -A myapp worker -Q email,processing -l info
# All queues
celery -A myapp worker -l info
# With priorities
celery -A myapp worker -Q high_priority,default -l info
# Exchange types
from kombu import Exchange, Queue
# Direct exchange (default)
direct_exchange = Exchange('direct', type='direct')
# Topic exchange (pattern matching)
topic_exchange = Exchange('topic', type='topic')
# Fanout exchange (broadcast)
fanout_exchange = Exchange('fanout', type='fanout')
app.conf.task_queues = (
Queue('tasks', exchange=direct_exchange, routing_key='tasks'),
Queue('logs', exchange=topic_exchange, routing_key='logs.*'),
Queue('broadcast', exchange=fanout_exchange),
)
# Dynamic routing
class TaskRouter:
def route_for_task(self, task, args=None, kwargs=None):
if task.startswith('email.'):
return {'queue': 'email'}
elif task.startswith('processing.'):
return {'queue': 'processing'}
return {'queue': 'default'}
app.conf.task_routes = (TaskRouter(),)
Workflows & Chains
Task Composition
# Chains - Execute tasks in sequence
from celery import chain
# Each task receives previous result
workflow = chain(
add.s(2, 2), # 4
add.s(4), # 4 + 4 = 8
add.s(8) # 8 + 8 = 16
)
result = workflow()
print(result.get()) # 16
# Alternative syntax
workflow = (
add.s(2, 2) |
add.s(4) |
add.s(8)
)
# Groups - Execute tasks in parallel
from celery import group
job = group(
add.s(2, 2),
add.s(4, 4),
add.s(8, 8)
)
result = job()
print(result.get()) # [4, 8, 16]
# Chords - Group + callback
from celery import chord
# Execute group, then callback with all results
workflow = chord(
group(
add.s(2, 2),
add.s(4, 4),
add.s(8, 8)
)
)(sum_results.s()) # sum_results receives [4, 8, 16]
result = workflow.get()
# Map - Apply same task to multiple inputs
from celery import group
# Process multiple items
items = [1, 2, 3, 4, 5]
job = group(process_item.s(item) for item in items)
result = job()
results = result.get()
# Starmap - Unpack arguments
items = [(2, 2), (4, 4), (8, 8)]
job = group(add.starmap(items))
result = job()
print(result.get()) # [4, 8, 16]
# Complex workflow
# Process → Multiple parallel tasks → Aggregate
workflow = (
fetch_data.s(data_id) |
group(
process_part.s('part1'),
process_part.s('part2'),
process_part.s('part3')
) |
aggregate_results.s()
)
result = workflow.apply_async()
# Conditional execution
from celery import chain
def build_workflow(use_cache=True):
if use_cache:
return chain(
get_from_cache.s(key) |
process.s()
)
else:
return chain(
fetch_from_db.s(key) |
process.s() |
save_to_cache.s()
)
workflow = build_workflow(use_cache=False)
result = workflow.apply_async(args=['my_key'])
# Error handling in chains
from celery import chain
@app.task
def handle_error(request, exc, traceback):
print(f"Error in workflow: {exc}")
# Cleanup or notification logic
workflow = chain(
task1.s(),
task2.s(),
task3.s()
).on_error(handle_error.s())
# Parallel branches with different callbacks
from celery import group, chain
branch1 = chain(
task_a.s(),
task_b.s()
)
branch2 = chain(
task_c.s(),
task_d.s()
)
workflow = group(branch1, branch2)
result = workflow()
# Dynamic workflow generation
def create_processing_workflow(items):
tasks = []
# Pre-process
tasks.append(initialize.s())
# Process each item
for item in items:
tasks.append(process_item.s(item))
# Post-process
tasks.append(finalize.s())
return chain(*tasks)
workflow = create_processing_workflow(['item1', 'item2', 'item3'])
result = workflow.apply_async()
# Nested workflows
inner_workflow = group(
task1.s(),
task2.s(),
task3.s()
)
outer_workflow = chain(
prepare.s(),
inner_workflow,
finalize.s()
)
# Workflow with partial results
@app.task
def collect_results(results):
successful = [r for r in results if r.get('status') == 'success']
failed = [r for r in results if r.get('status') == 'failed']
return {
'successful': len(successful),
'failed': len(failed),
'total': len(results)
}
workflow = chord(
group(
process.s(item) for item in items
)
)(collect_results.s())
# Signature immutability
# Immutable signature (doesn't pass result)
sig = task.si(args)
# Regular signature (passes result)
sig = task.s(args)
workflow = chain(
task1.s(),
task2.si(fixed_args), # Ignores result from task1
task3.s() # Receives result from task1
)
Error Handling & Retries
Retry Patterns
# Basic retry
@app.task(bind=True, max_retries=3)
def task_with_retry(self, data):
try:
result = process(data)
return result
except Exception as exc:
# Retry after 60 seconds
raise self.retry(exc=exc, countdown=60)
# Retry with exponential backoff
@app.task(bind=True, max_retries=5)
def task_with_backoff(self, url):
try:
response = requests.get(url)
response.raise_for_status()
return response.json()
except requests.RequestException as exc:
# 2^retry seconds
countdown = 2 ** self.request.retries
raise self.retry(exc=exc, countdown=countdown)
# Retry with max delay
@app.task(bind=True, max_retries=10)
def task_with_max_delay(self, data):
try:
return api_call(data)
except APIError as exc:
# Max 5 minutes between retries
countdown = min(60 * (2 ** self.request.retries), 300)
raise self.retry(exc=exc, countdown=countdown)
# Selective retry
@app.task(bind=True, autoretry_for=(ConnectionError, TimeoutError))
def task_auto_retry(self, url):
response = requests.get(url, timeout=10)
return response.json()
# Custom retry policy
@app.task(
bind=True,
autoretry_for=(Exception,),
retry_kwargs={'max_retries': 5},
retry_backoff=True, # Exponential backoff
retry_backoff_max=600, # Max 10 minutes
retry_jitter=True # Add randomness
)
def task_with_policy(self, data):
return process(data)
# Manual retry logic
@app.task(bind=True, max_retries=3)
def task_manual_retry(self, order_id):
try:
order = get_order(order_id)
if order.status == 'pending':
# Retry if not ready
raise self.retry(countdown=30)
return process_order(order)
except OrderNotFound:
# Don't retry for this error
raise
except Exception as exc:
raise self.retry(exc=exc)
# Retry with different queues
@app.task(bind=True, max_retries=3)
def task_retry_different_queue(self, data):
try:
return process(data)
except Exception as exc:
# Move to low priority queue on retry
raise self.retry(
exc=exc,
countdown=60,
queue='low_priority'
)
# Error callbacks
@app.task
def error_handler(request, exc, traceback):
logger.error(f"Task {request.id} failed: {exc}")
send_alert_email(f"Task failed: {exc}")
@app.task(bind=True, on_failure=error_handler)
def task_with_error_callback(self, data):
return process(data)
# Custom exceptions
class RetryableError(Exception):
pass
class PermanentError(Exception):
pass
@app.task(bind=True, max_retries=3)
def task_custom_exceptions(self, data):
try:
return process(data)
except RetryableError as exc:
# Retry these
raise self.retry(exc=exc, countdown=60)
except PermanentError:
# Don't retry these
logger.error("Permanent error, not retrying")
raise
# Ignore specific errors
@app.task(throws=(KeyError, ValueError))
def task_ignore_errors(data):
# These exceptions won't be logged as errors
if 'required_field' not in data:
raise KeyError('Missing required field')
return process(data)
# Dead letter queue pattern
@app.task(bind=True, max_retries=3)
def task_with_dlq(self, data):
try:
return process(data)
except Exception as exc:
if self.request.retries >= self.max_retries:
# Send to dead letter queue
dead_letter_queue.apply_async(
args=[self.request.id, data, str(exc)]
)
raise self.retry(exc=exc, countdown=60)
@app.task
def dead_letter_queue(task_id, data, error):
# Store failed task for manual inspection
FailedTask.objects.create(
task_id=task_id,
data=data,
error=error,
timestamp=datetime.utcnow()
)
# Alert administrators
notify_admins(f"Task {task_id} failed permanently")
# Circuit breaker pattern
from collections import defaultdict
from datetime import datetime, timedelta
circuit_breaker = defaultdict(lambda: {'failures': 0, 'last_failure': None})
@app.task(bind=True)
def task_with_circuit_breaker(self, service_name, data):
breaker = circuit_breaker[service_name]
# Check if circuit is open
if breaker['failures'] >= 5:
last_failure = breaker['last_failure']
if datetime.utcnow() - last_failure < timedelta(minutes=5):
raise Exception(f"Circuit breaker open for {service_name}")
try:
result = call_service(service_name, data)
# Reset on success
breaker['failures'] = 0
return result
except Exception as exc:
# Increment failure count
breaker['failures'] += 1
breaker['last_failure'] = datetime.utcnow()
raise self.retry(exc=exc, countdown=60)
Resources & Learning Path
Learning Progression
Phase 1: Celery Basics (1-2 weeks) □ Basic task definition and execution □ Task queues and workers □ Message brokers (Redis/RabbitMQ) □ Result backends □ Simple error handling Phase 2: Intermediate Celery (2-3 weeks) □ Periodic tasks with Beat □ Task routing and priorities □ Workflows (chains, groups, chords) □ Retry strategies □ Task monitoring Phase 3: Advanced Celery (3-4 weeks) □ Custom task classes □ Advanced routing patterns □ Performance optimization □ Scaling strategies □ Production deployment Phase 4: Production Celery (Ongoing) □ High availability setup □ Monitoring and alerting □ Security hardening □ Multi-broker configurations □ Disaster recovery
Related Comprehensive Sheets
Python Ecosystem → Python Advanced (async patterns) → Django Production (integration) → FastAPI Advanced (background tasks) → Flask Production (integration) Infrastructure → Redis Advanced → RabbitMQ → Message Queue Patterns → Distributed Systems Monitoring & Operations → Monitoring & Observability → Docker & Kubernetes → CI/CD Pipelines → Production Deployment Performance → Performance Optimization → Scalability Patterns → Load Testing → Database Optimization
Pro Tips Summary
Task Design ✓ Keep tasks small and focused ✓ Make tasks idempotent when possible ✓ Use proper error handling ✓ Implement retry strategies ✓ Add logging for debugging ✓ Use task signatures for workflows Performance ✓ Use connection pooling ✓ Optimize task routing ✓ Monitor queue lengths ✓ Scale workers horizontally ✓ Use appropriate serializers ✓ Implement task priorities Reliability ✓ Enable task acks_late ✓ Set task time limits ✓ Implement health checks ✓ Monitor worker status ✓ Use result expiration ✓ Handle broker failures Security ✓ Validate task inputs ✓ Use secure connections ✓ Limit task permissions ✓ Sanitize task results ✓ Monitor for anomalies ✓ Keep dependencies updated Operations ✓ Use separate queues for priorities ✓ Monitor task latency ✓ Set up alerting ✓ Keep workers updated ✓ Use supervisord or systemd ✓ Implement graceful shutdowns