celery · diff
v1.0.0 to v2.0.0
8 added, 801 removed. Audit A to A.
---
name: celery
- description: "Distributed task queue for Python supporting Redis, RabbitMQ, periodic scheduling, real-time processing, and Django integration"
+ description: Use when running background tasks with Celery - worker and broker configuration (Redis, RabbitMQ, SQS), task definitions, chains and chords, Celery Beat periodic tasks, routing, retries, Flower monitoring, or testing tasks with pytest
metadata:
author: mte90
- version: "1.0.0"
+ version: 2.0.0
tags:
- python
- task-queue
- async
- distributed
- celery
---
# Celery
Complete reference for distributed task processing with Celery.
## Django 6.0 Alternative
For complex Django projects, consider **Django 6.0's built-in task queue** as a simpler alternative to Celery. It provides:
- Native Django integration (no separate broker needed)
- Database-backed task storage
- Simpler deployment (single Django process)
- Standardized interface: `from django.task import task`
**When to choose Django Tasks over Celery:**
- Existing Django project on version 6.0+
- Simpler use cases (single server, moderate load)
- Preference for minimal dependencies
- Database as task storage is acceptable
**When to keep Celery:**
- Multi-server distributed deployment
- High-throughput requirements (1000s of tasks/sec)
- Multiple broker options (Redis, RabbitMQ, etc.)
- Advanced features (task prioritization, complex routing)
- Multi-language support
## Overview
Celery is a distributed task queue system for Python that enables asynchronous task execution, scheduled tasks, and real-time processing.
**Key Features:**
- Distributed: Run workers across multiple machines
- Brokers: Redis, RabbitMQ, SQS, and more
- Scheduled tasks: Periodic execution with Celery Beat
- Real-time: Task monitoring and result tracking
- Scalable: Easy horizontal scaling
### Architecture
```
Producer → Broker → Worker → Result Backend
↓ ↓ ↓ ↓
Task Queue Consumer Storage
```
### When to Use Celery
- Email sending
- Image/video processing
- Report generation
- Web scraping
- Scheduled maintenance tasks
- Long-running computations
- Third-party API calls
## Installation
```bash
pip install celery
# With Redis broker (recommended)
pip install celery[redis]
# With RabbitMQ broker
pip install celery[amqp]
# With Django
pip install django-celery-beat django-celery-results
```
### Basic Configuration
```python
# celery_config.py or celery.py
from celery import Celery
# Create app
app = Celery('myapp')
# Configure from Django settings
app.config_from_object('django.conf:settings', namespace='CELERY')
# Or configure directly
app.conf.update(
broker_url='redis://localhost:6379/0',
result_backend='redis://localhost:6379/0',
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='UTC',
enable_utc=True,
)
```
### Django Integration
```python
# myproject/celery.py
import os
from celery import Celery
# Set default Django settings
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
app = Celery('myproject')
# Load config from Django settings with CELERY_ prefix
app.config_from_object('django.conf:settings', namespace='CELERY')
# Auto-discover tasks in installed apps
app.autodiscover_tasks()
# myproject/__init__.py
from .celery import app as celery_app
__all__ = ('celery_app',)
# settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'UTC'
```
### Project Structure
```
myproject/
├── celery.py
├── __init__.py
├── settings.py
└── apps/
└── myapp/
├── tasks.py
└── __init__.py
```
## Brokers
### Redis
```python
# Basic Redis configuration
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
# Redis with authentication
CELERY_BROKER_URL = 'redis://:password@localhost:6379/0'
# Redis Sentinel for HA
CELERY_BROKER_URL = 'sentinel://localhost:26379/0;sentinel://localhost:26380/0;sentinel://localhost:26381/0'
CELERY_BROKER_TRANSPORT_OPTIONS = {
'master_name': 'mymaster',
}
# Redis with SSL
CELERY_BROKER_URL = 'rediss://localhost:6379/0'
CELERY_BROKER_TRANSPORT_OPTIONS = {
'ssl_cert_reqs': 'required',
}
```
### RabbitMQ
```python
# Basic RabbitMQ configuration
CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672//'
# RabbitMQ with authentication
CELERY_BROKER_URL = 'amqp://user:password@localhost:5672/vhost'
# RabbitMQ with SSL
CELERY_BROKER_URL = 'amqps://user:password@localhost:5671/vhost'
CELERY_BROKER_TRANSPORT_OPTIONS = {
'ssl_cert_reqs': 'required',
}
```
### Amazon SQS
```python
CELERY_BROKER_URL = 'sqs://aws_access_key:aws_secret_key@'
CELERY_BROKER_TRANSPORT_OPTIONS = {
'region': 'us-east-1',
'visibility_timeout': 3600,
'polling_interval': 1,
}
```
## Result Backends
### Redis
```python
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
# With expiration
CELERY_RESULT_EXPIRES = 3600 # 1 hour
```
### Database (Django)
```python
# Install: pip install django-celery-results
INSTALLED_APPS = [
# ...
'django_celery_results',
]
CELERY_RESULT_BACKEND = 'django-db'
CELERY_CACHE_BACKEND = 'django-cache'
```
### RPC
```python
# Results sent back as AMQP messages
CELERY_RESULT_BACKEND = 'rpc://'
```
## Task Definitions
### Basic Task
```python
from celery import shared_task
@shared_task
def add(x, y):
"""Simple addition task."""
return x + y
@shared_task
def send_email(to, subject, body):
"""Send email task."""
# Email sending logic
send_mail(subject, body, 'from@example.com', [to])
return f"Email sent to {to}"
```
### Task Options
```python
from celery import shared_task
@shared_task(
name='myapp.process_data', # Custom task name
bind=True, # Access to task instance
max_retries=3, # Max retry attempts
default_retry_delay=60, # Delay between retries (seconds)
autoretry_for=(Exception,), # Auto-retry on these exceptions
retry_backoff=True, # Exponential backoff
retry_backoff_max=600, # Max backoff delay
retry_jitter=True, # Add jitter to backoff
time_limit=300, # Hard time limit (seconds)
soft_time_limit=240, # Soft time limit
rate_limit='10/m', # Rate limiting
ignore_result=False, # Store result
store_errors_even_if_ignored=True,
)
def process_data(self, data_id):
"""Process data with full task options."""
try:
data = Data.objects.get(id=data_id)
result = data.process()
return result
except Data.DoesNotExist:
raise self.retry(countdown=60)
except Exception as exc:
raise self.retry(exc=exc, countdown=60)
```
### Bound Tasks
```python
@shared_task(bind=True)
def bound_task(self, x, y):
"""Task with access to task instance."""
# Access task ID
task_id = self.request.id
# Access retries count
retries = self.request.retries
# Update task state
self.update_state(
state='PROGRESS',
meta={'current': 50, 'total': 100}
)
# Retry manually
try:
return risky_operation(x, y)
except Exception as exc:
raise self.retry(exc=exc, countdown=5)
```
### Task Inheritance
```python
from celery import Task
class DatabaseTask(Task):
"""Custom task base class with database session."""
_db = None
@property
def db(self):
if self._db is None:
self._db = DatabaseSession()
return self._db
def after_return(self, *args, **kwargs):
if self._db is not None:
self._db.close()
self._db = None
@shared_task(base=DatabaseTask, bind=True)
def database_task(self, record_id):
"""Task using custom base class."""
record = self.db.query(Record).get(record_id)
return record.process()
```
### Task Signals
```python
from celery import signals
@signals.task_prerun.connect
def task_prerun_handler(sender=None, task_id=None, task=None, **kwargs):
"""Called before task execution."""
print(f"Task {task.name}[{task_id}] starting...")
@signals.task_postrun.connect
def task_postrun_handler(sender=None, task_id=None, task=None, retval=None, **kwargs):
"""Called after task execution."""
print(f"Task {task.name}[{task_id}] completed with result: {retval}")
@signals.task_failure.connect
def task_failure_handler(sender=None, task_id=None, exception=None, **kwargs):
"""Called on task failure."""
print(f"Task {task_id} failed: {exception}")
@signals.task_retry.connect
def task_retry_handler(sender=None, reason=None, **kwargs):
"""Called on task retry."""
print(f"Task retrying: {reason}")
```
- ## Calling Tasks
-
- ### Basic Calls
-
- ```python
- # Apply async (recommended)
- result = add.apply_async((2, 3))
-
- # Using delay (shortcut for apply_async)
- result = add.delay(2, 3)
-
- # Get result
- value = result.get() # Blocks until ready
- value = result.get(timeout=10) # With timeout
-
- # Check status
- result.ready() # True if completed
- result.successful() # True if successful
- result.failed() # True if failed
- result.state # PENDING, STARTED, SUCCESS, FAILURE, RETRY, REVOKED
- ```
-
- ### apply_async Options
-
- ```python
- # Countdown (delay in seconds)
- result = add.apply_async((2, 3), countdown=60)
-
- # ETA (specific time)
- from datetime import datetime, timedelta
- eta = datetime.utcnow() + timedelta(hours=1)
- result = add.apply_async((2, 3), eta=eta)
-
- # Expires
- result = add.apply_async((2, 3), expires=3600) # 1 hour
-
- # Specific queue
- result = add.apply_async((2, 3), queue='high_priority')
-
- # Specific worker
- result = add.apply_async((2, 3), worker='worker1@hostname')
-
- # Retry policy
- result = add.apply_async(
- (2, 3),
- retry=True,
- retry_policy={
- 'max_retries': 3,
- 'interval_start': 0,
- 'interval_step': 0.2,
- 'interval_max': 0.5,
- }
- )
-
- # Priority (0-255, higher = more important)
- result = add.apply_async((2, 3), priority=10)
-
- # Custom task ID
- result = add.apply_async((2, 3), task_id='custom-task-id')
-
- # Compression
- result = add.apply_async((large_data,), compression='gzip')
- ```
-
- ### Signatures
-
- ```python
- from celery import signature
-
- # Create signature
- s = add.s(2, 3)
- s.apply_async()
-
- # Partial signature (immutable)
- s = add.s(2) # First argument fixed
- s.apply_async(args=(3,)) # Will call add(2, 3)
-
- # Signature with options
- s = add.s(2, 3).set(countdown=10).set(queue='priority')
- s.apply_async()
-
- # Clone signature
- s2 = s.clone(args=(4, 5))
- ```
-
- ### Chains
-
- ```python
- from celery import chain
-
- # Sequential execution
- workflow = chain(add.s(2, 3), add.s(4), add.s(5))
- result = workflow.apply_async()
- # add(2, 3) → add(result, 4) → add(result, 5)
-
- # Pipe notation
- workflow = add.s(2, 3) | add.s(4) | add.s(5)
- result = workflow.apply_async()
-
- # With error handling
- workflow = chain(
- validate_data.s(data_id),
- process_data.s(),
- save_result.s()
- )
- result = workflow.apply_async()
- ```
-
- ### Groups
-
- ```python
- from celery import group
-
- # Parallel execution
- job = group(add.s(i, i) for i in range(10))
- result = job.apply_async()
-
- # Get all results
- values = result.get() # [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
-
- # Check completion
- result.ready() # True when all complete
- result.completed_count() # Number completed
-
- # Group with callback
- job = group(process_item.s(item_id) for item_id in items)
- callback = summarize_results.s()
- workflow = job | callback
- result = workflow.apply_async()
- ```
-
- ### Chords
-
- ```python
- from celery import chord
-
- # Group with callback (parallel + callback)
- callback = summarize.s()
- header = [process_item.s(item_id) for item_id in items]
- result = chord(header)(callback)
-
- # All header tasks run in parallel, then callback receives results
- @shared_task
- def summarize(results):
- return sum(results)
-
- # Example: Process order
- workflow = chord(
- group(
- validate_inventory.s(item_id)
- for item_id in order_items
- ),
- create_shipment.s(order_id)
- )
- result = workflow.apply_async()
- ```
-
- ### Chunks
-
- ```python
- # Split large task into chunks
- @shared_task
- def process_chunk(items):
- return [process_item(item) for item in items]
-
- # Process 100 items in chunks of 10
- items = list(range(100))
- result = process_chunk.chunks(items, 10).apply_async()
- ```
-
- ## Task States
-
- ### Built-in States
-
- ```python
- from celery.result import AsyncResult
-
- result = add.delay(2, 3)
-
- # States
- result.state
- # PENDING - Task waiting to execute
- # STARTED - Task started
- # SUCCESS - Task completed successfully
- # FAILURE - Task failed
- # RETRY - Task being retried
- # REVOKED - Task revoked
-
- # Check state
- if result.successful():
- print(f"Result: {result.result}")
- elif result.failed():
- print(f"Error: {result.result}")
- ```
-
- ### Custom States
-
- ```python
- @shared_task(bind=True)
- def long_task(self, total):
- """Task with progress tracking."""
- for i in range(total):
- # Update progress
- self.update_state(
- state='PROGRESS',
- meta={
- 'current': i + 1,
- 'total': total,
- 'status': 'Processing...'
- }
- )
- time.sleep(1)
-
- return {'result': 'completed'}
-
- # Check custom state
- result = long_task.delay(10)
- if result.state == 'PROGRESS':
- progress = result.info
- print(f"{progress['current']}/{progress['total']}")
- ```
-
- ## Periodic Tasks (Celery Beat)
-
- ### Configuration
-
- ```python
- # celery.py
- from celery.schedules import crontab
-
- app.conf.beat_schedule = {
- # Every 30 seconds
- 'add-every-30-seconds': {
- 'task': 'myapp.tasks.add',
- 'schedule': 30.0,
- 'args': (16, 16)
- },
-
- # Crontab schedule
- 'cleanup-every-night': {
- 'task': 'myapp.tasks.cleanup',
- 'schedule': crontab(hour=2, minute=0),
- },
-
- # Every Monday morning
- 'weekly-report': {
- 'task': 'myapp.tasks.weekly_report',
- 'schedule': crontab(hour=7, minute=30, day_of_week=1),
- },
-
- # Every 5 minutes during business hours
- 'business-check': {
- 'task': 'myapp.tasks.check_status',
- 'schedule': crontab(minute='*/5', hour='9-17'),
- },
- }
-
- app.conf.timezone = 'UTC'
- ```
-
- ### Crontab Syntax
-
- ```python
- from celery.schedules import crontab
-
- # Every minute
- crontab()
-
- # Every hour at minute 0
- crontab(minute=0)
-
- # Every day at midnight
- crontab(hour=0, minute=0)
-
- # Every Monday at 8:30 AM
- crontab(hour=8, minute=30, day_of_week=1)
-
- # Every 15 minutes
- crontab(minute='*/15')
-
- # First day of every month
- crontab(hour=0, minute=0, day_of_month=1)
-
- # Every weekday (Mon-Fri) at 6 PM
- crontab(hour=18, minute=0, day_of_week='1-5')
-
- # Specific months
- crontab(hour=0, minute=0, day_of_month=1, month_of_year='1,4,7,10')
- ```
-
- ### django-celery-beat
-
- ```python
- # Install: pip install django-celery-beat
-
- # settings.py
- INSTALLED_APPS = [
- # ...
- 'django_celery_beat',
- ]
-
- # Use database-backed schedule
- CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
-
- # Now manage periodic tasks via Django admin
- ```
-
- ```python
- # Create periodic task programmatically
- from django_celery_beat.models import PeriodicTask, IntervalSchedule
-
- # Create interval
- schedule, _ = IntervalSchedule.objects.get_or_create(
- every=10,
- period=IntervalSchedule.SECONDS,
- )
-
- # Create task
- PeriodicTask.objects.create(
- interval=schedule,
- name='my-periodic-task',
- task='myapp.tasks.my_task',
- args=json.dumps(['arg1', 'arg2']),
- kwargs=json.dumps({'key': 'value'}),
- )
-
- # Crontab schedule
- from django_celery_beat.models import CrontabSchedule
-
- schedule, _ = CrontabSchedule.objects.get_or_create(
- minute='0',
- hour='*',
- day_of_week='*',
- day_of_month='*',
- month_of_year='*',
- )
-
- PeriodicTask.objects.create(
- crontab=schedule,
- name='hourly-task',
- task='myapp.tasks.hourly',
- )
- ```
-
- ## Task Routing
-
- ### Queue Routing
-
- ```python
- # settings.py
- CELERY_TASK_ROUTES = {
- 'myapp.tasks.send_email': {
- 'queue': 'email',
- },
- 'myapp.tasks.process_video': {
- 'queue': 'video',
- 'routing_key': 'video.process',
- },
- 'myapp.tasks.*': {
- 'queue': 'default',
- },
- }
-
- # Or use task decorator
- @shared_task(queue='email')
- def send_email(to, subject, body):
- pass
-
- # Or specify when calling
- send_email.apply_async(args=[], queue='priority')
- ```
-
- ### Automatic Routing
-
- ```python
- # celery.py
- def route_task(name, args, kwargs, options, task=None, **kw):
- """Custom routing function."""
- if name.startswith('myapp.email.'):
- return {'queue': 'email'}
- if name.startswith('myapp.video.'):
- return {'queue': 'video', 'routing_key': 'video.high'}
- return {'queue': 'default'}
-
- app.conf.task_routes = (route_task,)
- ```
-
- ### Worker Queues
-
- ```bash
- # Start worker for specific queues
- celery -A myproject worker -Q email,video -n worker1@%h
-
- # Multiple workers for different queues
- celery -A myproject worker -Q email -n email_worker@%h --concurrency=4
- celery -A myproject worker -Q video -n video_worker@%h --concurrency=2
- celery -A myproject worker -Q default -n default_worker@%h
- ```
-
- ## Error Handling
-
- ### Retry Strategies
-
- ```python
- @shared_task(
- bind=True,
- max_retries=5,
- autoretry_for=(ConnectionError, TimeoutError),
- retry_backoff=True,
- retry_backoff_max=600,
- retry_jitter=True,
- )
- def fetch_external_api(self, url):
- """Fetch from external API with retry."""
- try:
- response = requests.get(url, timeout=10)
- response.raise_for_status()
- return response.json()
- except requests.RequestException as exc:
- raise self.retry(exc=exc)
-
- # Manual retry with custom logic
- @shared_task(bind=True, max_retries=3)
- def process_with_retry(self, data_id):
- try:
- data = Data.objects.get(id=data_id)
- return data.process()
- except Data.DoesNotExist:
- # Don't retry - data doesn't exist
- raise
- except DatabaseError as exc:
- # Retry with exponential backoff
- countdown = 2 ** self.request.retries
- raise self.retry(exc=exc, countdown=countdown)
- ```
-
- ### Error Callbacks
-
- ```python
- @shared_task
- def on_failure(request, exc, traceback):
- """Error callback task."""
- logger.error(f"Task {request.id} failed: {exc}")
- notify_admins(f"Task failed: {request.id}")
-
- @shared_task
- def process_data(data_id):
- data = Data.objects.get(id=data_id)
- return data.process()
-
- # Link error callback
- result = process_data.apply_async(
- args=[data_id],
- link_error=on_failure.s()
- )
- ```
-
- ### Dead Letter Queue
-
- ```python
- # Configure dead letter queue
- CELERY_TASK_ANNOTATIONS = {
- '*': {
- 'on_failure': handle_task_failure,
- }
- }
-
- @shared_task
- def handle_task_failure(task_id, exception, args, kwargs, traceback, einfo):
- """Handle failed tasks."""
- FailedTask.objects.create(
- task_id=task_id,
- exception=str(exception),
- args=args,
- kwargs=kwargs,
- traceback=traceback,
- )
- notify_admins(f"Task {task_id} failed permanently")
- ```
-
- ## Monitoring
-
- ### Flower (Web UI)
-
- ```bash
- # Install
- pip install flower
-
- # Start
- celery -A myproject flower --port=5555
-
- # With authentication
- celery -A myproject flower --port=5555 --basic-auth=user:password
- ```
-
- ```python
- # Configure in settings
- CELERY_FLOWER_PORT = 5555
- CELERY_FLOWER_BASIC_AUTH = ['user:password']
- ```
-
- ### Command Line Monitoring
-
- ```bash
- # Inspect active tasks
- celery -A myproject inspect active
-
- # Inspect registered tasks
- celery -A myproject inspect registered
-
- # Inspect scheduled tasks
- celery -A myproject inspect scheduled
-
- # Inspect reserved tasks
- celery -A myproject inspect reserved
-
- # Stats
- celery -A myproject inspect stats
-
- # Ping workers
- celery -A myproject inspect ping
-
- # Control workers
- celery -A myproject control enable_events
- celery -A myproject control disable_events
-
- # Revoke task
- celery -A myproject revoke <task_id>
-
- # Terminate task (SIGTERM)
- celery -A myproject revoke <task_id> --terminate
-
- # Purge all tasks
- celery -A myproject purge
- ```
-
- ### Prometheus Metrics
-
- ```python
- # Install: pip install celery-prometheus-exporter
-
- # Start exporter
- celery-prometheus-exporter --broker redis://localhost:6379/0
-
- # Or integrate into Django
- # urls.py
- from django.urls import path
- from celery_prometheus.views import metrics_view
-
- urlpatterns = [
- path('metrics/', metrics_view),
- ]
- ```
-
- ## Testing
-
- ### pytest Configuration
-
- ```python
- # conftest.py
- import pytest
- from celery import Celery
-
- @pytest.fixture(scope='session')
- def celery_app():
- """Create test Celery app."""
- app = Celery('test_app')
- app.config_from_object({
- 'broker_url': 'memory://',
- 'result_backend': 'cache+memory://',
- 'task_always_eager': True, # Execute synchronously
- 'task_eager_propagates': True, # Propagate exceptions
- })
- return app
-
- @pytest.fixture
- def celery_worker(celery_app):
- """Create test worker."""
- from celery.contrib.testing import worker
- with worker.start_worker(celery_app):
- yield celery_app
- ```
-
- ### Unit Tests
-
- ```python
- # tests/test_tasks.py
- import pytest
- from myapp.tasks import add, send_email
-
- # Synchronous testing (eager mode)
- @pytest.mark.celery(task_always_eager=True)
- def test_add_task(celery_app):
- """Test add task executes synchronously."""
- result = add.delay(2, 3)
- assert result.get() == 5
-
- # Mock task
- def test_send_email_task(mocker):
- """Test send_email task with mocked send."""
- mock_send = mocker.patch('myapp.tasks.send_mail')
-
- send_email.delay('to@example.com', 'Subject', 'Body')
-
- mock_send.assert_called_once_with(
- 'Subject', 'Body', 'from@example.com', ['to@example.com']
- )
-
- # Test with fixtures
- @pytest.fixture
- def mock_email_backend():
- """Mock email backend."""
- with patch('myapp.tasks.send_mail') as mock:
- yield mock
-
- def test_email_task_with_fixture(mock_email_backend):
- send_email.delay('test@example.com', 'Test', 'Body')
- assert mock_email_backend.called
- ```
-
- ### Integration Tests
-
- ```python
- # tests/test_integration.py
- import pytest
- from celery.result import AsyncResult
-
- @pytest.mark.integration
- def test_task_execution(celery_worker):
- """Test real task execution with worker."""
- result = add.apply_async(args=(10, 20))
-
- # Wait for result
- value = result.get(timeout=10)
-
- assert value == 30
- assert result.successful()
-
- @pytest.mark.integration
- def test_task_retry(celery_worker):
- """Test task retry behavior."""
- from myapp.tasks import flaky_task
-
- result = flaky_task.apply_async()
-
- # Should eventually succeed after retries
- value = result.get(timeout=30)
- assert value is not None
- ```
-
- ## Performance
-
- ### Concurrency
-
- ```python
- # Worker configuration
- CELERY_WORKER_CONCURRENCY = 4 # Number of worker processes
-
- # Prefetch limit
- CELERY_WORKER_PREFETCH_MULTIPLIER = 4 # Tasks per worker
-
- # For long tasks, reduce prefetch
- CELERY_WORKER_PREFETCH_MULTIPLIER = 1
- ```
-
- ```bash
- # Start with specific concurrency
- celery -A myproject worker --concurrency=4
-
- # Use eventlet for I/O-bound tasks
- celery -A myproject worker --pool=eventlet --concurrency=100
-
- # Use gevent
- celery -A myproject worker --pool=gevent --concurrency=100
- ```
-
- ### Task Optimization
-
- ```python
- # Avoid database queries in loops
- @shared_task
- def process_items_bad(item_ids):
- """Bad: N database queries."""
- results = []
- for item_id in item_ids:
- item = Item.objects.get(id=item_id) # N queries!
- results.append(item.process())
- return results
-
- @shared_task
- def process_items_good(item_ids):
- """Good: 1 database query."""
- items = Item.objects.filter(id__in=item_ids)
- return [item.process() for item in items]
-
- # Use bulk operations
- @shared_task
- def bulk_update_items(updates):
- """Use bulk_update for efficiency."""
- items = []
- for item_id, data in updates:
- item = Item(id=item_id, **data)
- items.append(item)
-
- Item.objects.bulk_update(
- items,
- ['field1', 'field2', 'field3']
- )
- ```
-
- ### Memory Management
-
- ```python
- # Process large datasets in chunks
- @shared_task
- def process_large_dataset(dataset_id):
- """Process large dataset without loading all into memory."""
- dataset = Dataset.objects.get(id=dataset_id)
-
- # Use iterator to avoid loading all records
- for batch in dataset.records.iterator(chunk_size=1000):
- process_batch(batch)
-
- # Configure max tasks per worker
- CELERY_WORKER_MAX_TASKS_PER_CHILD = 1000 # Restart after 1000 tasks
-
- # Memory limit
- CELERY_WORKER_MAX_MEMORY_PER_CHILD = 400000 # 400MB
- ```
-
## Best Practices
### 1. Idempotent Tasks
```python
@shared_task(bind=True)
def process_payment(self, payment_id):
"""Idempotent payment processing."""
payment = Payment.objects.select_for_update().get(id=payment_id)
# Check if already processed
if payment.status == 'completed':
return {'status': 'already_processed'}
# Process only once
with transaction.atomic():
result = payment.charge()
payment.mark_completed()
return result
```
### 2. Proper Error Handling
```python
@shared_task(bind=True, autoretry_for=(Exception,), max_retries=3)
def robust_task(self, data_id):
"""Task with proper error handling."""
try:
data = Data.objects.get(id=data_id)
except Data.DoesNotExist:
# Log and don't retry
logger.error(f"Data {data_id} not found")
return None
try:
return data.process()
except ExternalAPIError as exc:
# Retry with backoff
raise self.retry(exc=exc, countdown=2 ** self.request.retries)
except Exception as exc:
# Unexpected error - log and retry
logger.exception(f"Unexpected error processing {data_id}")
raise
```
### 3. Task Granularity
```python
# Bad: One monolithic task
@shared_task
def process_order_bad(order_id):
order = Order.objects.get(id=order_id)
order.validate()
order.charge()
order.ship()
order.send_confirmation()
# Good: Smaller, focused tasks
@shared_task
def validate_order(order_id):
order = Order.objects.get(id=order_id)
order.validate()
charge_order.delay(order_id)
@shared_task
def charge_order(order_id):
order = Order.objects.get(id=order_id)
order.charge()
ship_order.delay(order_id)
@shared_task
def ship_order(order_id):
order = Order.objects.get(id=order_id)
order.ship()
send_confirmation.delay(order_id)
```
### 4. Use Task Queues Appropriately
```python
# Route different task types to different queues
@shared_task(queue='high_priority')
def send_password_reset(user_id):
"""Time-sensitive task."""
pass
@shared_task(queue='low_priority')
def generate_report(user_id):
"""Background task."""
pass
@shared_task(queue='compute')
def process_video(video_id):
"""CPU-intensive task."""
pass
```
- ## Common Issues
-
- ### Issue: Tasks Stuck in PENDING
-
- **Problem**: Tasks never execute.
-
- **Solution**:
- ```bash
- # Check worker is running
- celery -A myproject inspect active
-
- # Check worker is consuming from correct queue
- celery -A myproject worker -Q myqueue
-
- # Check broker connection
- celery -A myproject inspect ping
- ```
-
- ### Issue: Memory Leaks
-
- **Problem**: Worker memory grows over time.
-
- **Solution**:
- ```python
- # Limit tasks per worker
- CELERY_WORKER_MAX_TASKS_PER_CHILD = 1000
-
- # Or limit memory
- CELERY_WORKER_MAX_MEMORY_PER_CHILD = 400000 # KB
-
- # Avoid circular references in tasks
- @shared_task
- def task_with_cleanup():
- try:
- return process()
- finally:
- # Clean up resources
- cleanup()
- ```
-
- ### Issue: Duplicate Task Execution
-
- **Problem**: Task executed multiple times.
+ ## Deep Dives
- **Solution**:
- ```python
- # Use idempotent tasks
- @shared_task(bind=True)
- def idempotent_task(self, data_id):
- with transaction.atomic():
- # Lock record
- data = Data.objects.select_for_update().get(id=data_id)
-
- if data.processed:
- return {'status': 'already_processed'}
-
- result = data.process()
- data.processed = True
- data.save()
-
- return result
+ For detailed reference on specific topics, load these on-demand:
- # Or use task deduplication
- CELERY_TASK_ANNOTATIONS = {
- 'myapp.tasks.*': {
- 'acks_late': True,
- 'reject_on_worker_lost': True,
- }
- }
- ```
+ - **Calling Workflows** (`references/calling-workflows.md`) — Task invocation patterns: `apply_async`, signatures, chains, groups, chords, chunks, and task states
+ - **Periodic Tasks & Routing** (`references/beat-routing.md`) — Celery Beat scheduling, crontab syntax, django-celery-beat, queue routing
+ - **Error Handling & Monitoring** (`references/errors-monitoring.md`) — Retries, error callbacks, DLQ, Flower, CLI monitoring, Prometheus
+ - **Testing & Performance** (`references/testing-performance.md`) — pytest testing, worker configuration, task optimization, memory management, common issues
## References
- **Official Documentation**: https://docs.celeryq.dev/
- **GitHub Repository**: https://github.com/celery/celery
- **Flower Monitoring**: https://github.com/mher/flower
- **Django Celery**: https://github.com/celery/django-celery
-