celery · diff
v2.0.0 to v2.0.0
362 added, 317 removed. Audit A to A.
---
name: celery
- 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
+ description: Use when running background tasks with Celery - worker and broker configuration (Redis, RabbitMQ), task routing by name vs queue, chains/groups/chords, retry patterns (autoretry_for, retry_backoff), acks_late semantics, failure detection, and monitoring with Flower
metadata:
author: mte90
version: 2.0.0
tags:
- python
- task-queue
- async
- distributed
- celery
---
# Celery
- Complete reference for distributed task processing with Celery.
+ Distributed task processing with focus on routing, composition, and failure handling.
- ## Django 6.0 Alternative
+ ## When Not to Use Celery
- For complex Django projects, consider **Django 6.0's built-in task queue** as a simpler alternative to Celery. It provides:
+ Before adopting Celery, check if you actually need it:
- - Native Django integration (no separate broker needed)
- - Database-backed task storage
- - Simpler deployment (single Django process)
- - Standardized interface: `from django.task import task`
+ | Scenario | Better Alternative |
+ |----------|-------------------|
+ | Short-lived fan-out (5-10 parallel HTTP calls) | Simple thread pool or `asyncio.gather` |
+ | Single-server Django app with moderate load | Django 6.0+ built-in task queue |
+ | Work dominated by long IO (10+ min tasks) | Task-per-request model or job queue (dramatiq, huey) |
+ | Operational overhead unacceptable | django-tasks, RQ, or simple database polling |
- **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
+ **Check before adopting:**
+ - Do you need multiple workers across machines? (If no → simpler queue)
+ - Do you have 100+ tasks/minute sustained? (If no → overkill)
+ - Can you tolerate operational cost (broker + worker fleet + monitoring)? (If no → huey/dramatiq)
- **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
+ ## Task Routing and Queue Configuration
- ## Overview
+ ### Routing by Task Name (Recommended)
- Celery is a distributed task queue system for Python that enables asynchronous task execution, scheduled tasks, and real-time processing.
+ ```python
+ # settings.py or celery.py
+ CELERY_TASK_ROUTES = {
+ 'myapp.tasks.send_email': {'queue': 'email'},
+ 'myapp.tasks.process_video': {'queue': 'video'},
+ 'myapp.tasks.*': {'queue': 'default'}, # Catch-all
+ }
+ ```
- **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
+ **Why routing by task name beats routing by queue name:**
+ - Task name is stable; queue name is an implementation detail
+ - You can move a task to a different queue without changing call sites
+ - Clearer intent: "send_email goes to email queue" vs "this call goes to high_priority"
- ### Architecture
+ ### Precedence: Route vs Queue Argument
+ ```python
+ # Task decorator sets default queue
+ @shared_task(queue='default')
+ def send_email(user_id):
+ pass
+
+ # apply_async queue overrides decorator
+ send_email.apply_async(args=[1], queue='high_priority') # Uses 'high_priority'
+
+ # But task_routes overrides both if configured
+ # CELERY_TASK_ROUTES = {'myapp.tasks.send_email': {'queue': 'critical'}}
+ # Result: 'critical' wins (routing function is final)
```
- Producer → Broker → Worker → Result Backend
- ↓ ↓ ↓ ↓
- Task Queue Consumer Storage
+
+ **Precedence order:** `task_routes` (routing function) > `apply_async(queue=...)` > decorator `@shared_task(queue=...)`
+
+ ### Per-Queue Worker Startup
+
+ ```bash
+ # High-priority queue: solo pool for immediate processing
+ celery -A myproject worker -Q high -P solo -n high@%h
+
+ # Low-priority queue: prefork for throughput
+ celery -A myproject worker -Q low -P prefork --concurrency=4 -n low@%h
+
+ # Mixed queues: one worker handles both
+ celery -A myproject worker -Q high,low -P prefork --concurrency=2
```
- ### When to Use Celery
+ **Pool choices:**
+ - `solo`: Single process, no concurrency. Use for debugging or tasks that can't be parallelized.
+ - `prefork`: Default. Good for CPU-bound or mixed workloads.
+ - `eventlet`/`gevent`: High concurrency for I/O-bound tasks (100+ simultaneous).
- - Email sending
- - Image/video processing
- - Report generation
- - Web scraping
- - Scheduled maintenance tasks
- - Long-running computations
- - Third-party API calls
+ ### Critical Failure: Route to Queue with No Worker
- ## Installation
+ ```python
+ # Task routed to queue no one consumes
+ CELERY_TASK_ROUTES = {'myapp.tasks.cleanup': {'queue': 'cleanup'}}
- ```bash
- pip install celery
+ # Problem: No worker started with -Q cleanup
+ # Result: Task sits in broker forever, silently. No error, no retry.
+ ```
- # With Redis broker (recommended)
- pip install celery[redis]
+ **Detection:**
+ ```bash
+ # Check which queues have active workers
+ celery -A myproject inspect active
- # With RabbitMQ broker
- pip install celery[amqp]
+ # Check queue depth
+ redis-cli LLEN celery # For Redis broker
- # With Django
- pip install django-celery-beat django-celery-results
+ # Flower UI: Monitor queue lengths per queue
```
- ### Basic Configuration
+ **Fix:** Start a worker for the queue OR remove the route so tasks go to a consumed queue.
+ ## Task Composition: Chains, Groups, Chords
+
+ ### When to Use Each
+
+ | Pattern | Use Case | Requires Backend |
+ |---------|----------|------------------|
+ | `chain` | Sequential steps (A → B → C) | No (but needed for `.get()`) |
+ | `group` | Parallel independent tasks | No (but needed for results) |
+ | `chord` | Parallel + callback (wait for all, then run) | **Yes** (stores intermediate results) |
+
+ ### Chains (Sequential)
+
```python
- # celery_config.py or celery.py
- from celery import Celery
+ from celery import chain
- # Create app
- app = Celery('myapp')
+ # Correct: Results flow through
+ workflow = chain(
+ validate_user.s(user_id),
+ process_data.s(),
+ send_notification.s()
+ )
+ result = workflow.apply_async()
+ # Each task's output becomes next task's input
- # Configure from Django settings
- app.config_from_object('django.conf:settings', namespace='CELERY')
+ # Error handling: link error callback
+ from celery import group
- # 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,
+ error_handler = handle_error.s()
+ workflow = (
+ chain(validate.s(), process.s()) |
+ save_result.s()
)
+ workflow.apply_async(link_error=error_handler)
```
- ### Django Integration
+ ### Groups (Parallel)
```python
- # myproject/celery.py
- import os
- from celery import Celery
+ from celery import group
- # Set default Django settings
- os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
+ # Correct: Independent parallel tasks
+ job = group(
+ fetch_data.s(source_id)
+ for source_id in [1, 2, 3, 4, 5]
+ )
+ result = job.apply_async()
+ all_results = result.get() # [res1, res2, res3, res4, res5]
- app = Celery('myproject')
+ # Wrong: Passing unbounded group into chord (deadlock risk)
+ # See chord section below
+ ```
- # Load config from Django settings with CELERY_ prefix
- app.config_from_object('django.conf:settings', namespace='CELERY')
+ ### Chords (Parallel + Callback)
- # Auto-discover tasks in installed apps
- app.autodiscover_tasks()
+ ```python
+ from celery import chord, group
- # myproject/__init__.py
- from .celery import app as celery_app
+ # Correct form: Bounded group as chord header
+ result = chord(
+ group(
+ process_item.s(item_id)
+ for item_id in item_ids[:100] # Bounded list
+ ),
+ summarize_results.s() # Callback receives [res1, res2, ...]
+ ).apply_async()
- __all__ = ('celery_app',)
+ # Incorrect form that hangs: Unbounded generator
+ # DO NOT DO THIS:
+ # result = chord(
+ # group(process_item.s(item_id) for item_id in infinite_generator()),
+ # callback.s()
+ # )
+ # Why: Chord waits for ALL header tasks to complete. Generator never ends.
- # 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'
+ # Also incorrect: Group with dynamic size > worker capacity
+ # If you have 4 workers but group(1000 tasks), some tasks wait.
+ # Chord callback never fires until all 1000 complete.
+ # Solution: Use chunks or multiple chords
```
- ### Project Structure
+ **Chord backend requirement:**
+ ```python
+ # settings.py
+ CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' # Required for chords
- ```
- myproject/
- ├── celery.py
- ├── __init__.py
- ├── settings.py
- └── apps/
- └── myapp/
- ├── tasks.py
- └── __init__.py
+ # Without backend, chord callback never fires
+ # Error: Chord 'xxx' raised: 'NoneType' object is not subscriptable
```
- ## Brokers
+ ## Retry and Failure Handling
- ### Redis
+ ### Auto-Retry Configuration
```python
- # Basic Redis configuration
- CELERY_BROKER_URL = 'redis://localhost:6379/0'
- CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
+ @shared_task(
+ bind=True,
+ autoretry_for=(ConnectionError, TimeoutError, ExternalAPIError),
+ retry_backoff=True, # Exponential backoff (1s, 2s, 4s, 8s...)
+ retry_backoff_max=600, # Cap at 10 minutes
+ retry_jitter=True, # Add randomness to prevent thundering herd
+ max_retries=3,
+ )
+ def fetch_external_data(self, url):
+ """Auto-retry on transient failures."""
+ response = requests.get(url, timeout=10)
+ response.raise_for_status()
+ return response.json()
+ ```
- # Redis with authentication
- CELERY_BROKER_URL = 'redis://:password@localhost:6379/0'
+ **Retry parameters explained:**
+ - `autoretry_for`: Tuple of exception classes that trigger auto-retry
+ - `retry_backoff=True`: Delays grow exponentially (default: 1s to 60s)
+ - `retry_backoff_max`: Maximum delay between retries
+ - `retry_jitter=True`: Add ±25% randomness to each delay
+ - `max_retries`: Total retry attempts (excluding first execution)
- # 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',
- }
+ ### Manual Retry
- # Redis with SSL
- CELERY_BROKER_URL = 'rediss://localhost:6379/0'
- CELERY_BROKER_TRANSPORT_OPTIONS = {
- 'ssl_cert_reqs': 'required',
- }
+ ```python
+ @shared_task(bind=True, max_retries=5)
+ def process_with_manual_retry(self, record_id):
+ try:
+ record = Record.objects.get(id=record_id)
+ return record.process()
+ except Record.DoesNotExist:
+ # Don't retry: data missing permanently
+ logger.error(f"Record {record_id} not found")
+ return None
+ except DatabaseError as exc:
+ # Retry with custom countdown
+ countdown = min(60 * (2 ** self.request.retries), 300)
+ raise self.retry(exc=exc, countdown=countdown)
+ except Exception as exc:
+ # Unexpected error: re-raise without retry
+ logger.exception("Unexpected error")
+ raise
```
- ### RabbitMQ
+ ### What NOT to Retry
```python
- # Basic RabbitMQ configuration
- CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672//'
-
- # RabbitMQ with authentication
- CELERY_BROKER_URL = 'amqp://user:password@localhost:5672/vhost'
+ # Non-idempotent side effects: Don't retry
+ @shared_task(bind=True, max_retries=3)
+ def charge_credit_card(self, payment_id):
+ """WRONG: Charging card multiple times on retry."""
+ payment = Payment.objects.get(id=payment_id)
+ stripe.Charge.create(amount=payment.amount, card=payment.card_id)
+ # If worker crashes after charge but before DB update, retry charges again
- # RabbitMQ with SSL
- CELERY_BROKER_URL = 'amqps://user:password@localhost:5671/vhost'
- CELERY_BROKER_TRANSPORT_OPTIONS = {
- 'ssl_cert_reqs': 'required',
- }
+ # Correct: Make idempotent
+ @shared_task(bind=True, max_retries=3)
+ def charge_credit_card_idempotent(self, payment_id):
+ payment = Payment.objects.select_for_update().get(id=payment_id)
+
+ if payment.status == 'charged':
+ return {'status': 'already_charged'}
+
+ with transaction.atomic():
+ payment.refresh_from_db()
+ if payment.status == 'charged':
+ return {'status': 'already_charged'}
+
+ stripe.Charge.create(amount=payment.amount, card=payment.card_id)
+ payment.status = 'charged'
+ payment.save()
+
+ return {'status': 'charged'}
```
- ### Amazon SQS
+ ### acks_late and task_reject_on_worker_lost
```python
- CELERY_BROKER_URL = 'sqs://aws_access_key:aws_secret_key@'
+ # settings.py
+ CELERY_ACKS_LATE = True # Acknowledge task AFTER execution completes
- CELERY_BROKER_TRANSPORT_OPTIONS = {
- 'region': 'us-east-1',
- 'visibility_timeout': 3600,
- 'polling_interval': 1,
- }
+ CELERY_TASK_REJECT_ON_WORKER_LOST = True # Reject unacked tasks when worker dies
```
- ## Result Backends
+ **Semantics and trade-offs:**
- ### Redis
+ | Setting | Behavior | Trade-off |
+ |---------|----------|-----------|
+ | `acks_late=False` (default) | Ack immediately on receipt | Worker crash = task lost (no retry) |
+ | `acks_late=True` | Ack after task completes | Worker crash = task requeued (may run twice) |
+ | `reject_on_worker_lost=True` | Reject unacked tasks on worker death | Same as acks_late but explicit |
+ **At-least-once delivery means idempotency is your responsibility:**
```python
- CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
-
- # With expiration
- CELERY_RESULT_EXPIRES = 3600 # 1 hour
+ # With acks_late=True, this task may run twice:
+ @shared_task(acks_late=True)
+ def send_welcome_email(user_id):
+ # WRONG: Sends duplicate email
+ User.objects.get(id=user_id).send_email()
+
+ # Correct: Idempotent
+ @shared_task(acks_late=True)
+ def send_welcome_email_idempotent(user_id):
+ user = User.objects.get(id=user_id)
+ if user.welcome_email_sent:
+ return {'status': 'already_sent'}
+
+ with transaction.atomic():
+ user.refresh_from_db()
+ if user.welcome_email_sent:
+ return {'status': 'already_sent'}
+ user.send_email()
+ user.welcome_email_sent = True
+ user.save()
```
- ### Database (Django)
+ ## Common Issues (Symptom-First)
- ```python
- # Install: pip install django-celery-results
- INSTALLED_APPS = [
- # ...
- 'django_celery_results',
- ]
+ ### Symptom: Tasks Never Execute (Stuck in PENDING)
- CELERY_RESULT_BACKEND = 'django-db'
- CELERY_CACHE_BACKEND = 'django-cache'
- ```
+ **How to confirm:**
+ ```bash
+ # 1. Check worker is running and consuming correct queue
+ celery -A myproject inspect active
- ### RPC
+ # 2. Check queue routing
+ python -c "from myproject.celery import app; print(app.conf.task_routes)"
- ```python
- # Results sent back as AMQP messages
- CELERY_RESULT_BACKEND = 'rpc://'
+ # 3. Check broker queue depth
+ redis-cli LLEN celery # Should decrease as tasks execute
+
+ # 4. Verify task name matches
+ celery -A myproject inspect registered # Task must appear here
```
- ## Task Definitions
+ **Fix:**
+ - Queue misrouting: Add route or start worker with `-Q <queue_name>`
+ - Missing `default_queue`: Set `CELERY_TASK_DEFAULT_QUEUE = 'default'`
+ - Task not registered: Check `autodiscover_tasks()` is called
- ### Basic Task
+ ### Symptom: Tasks Silently Lost (No Error, No Execution)
- ```python
- from celery import shared_task
+ **How to confirm:**
+ ```bash
+ # 1. Check broker/result backend mismatch
+ # If broker=redis://localhost:6379/0 but result_backend=redis://localhost:6379/1
+ # Results go to different Redis instance
- @shared_task
- def add(x, y):
- """Simple addition task."""
- return x + y
+ # 2. Check worker logs for "Task xxx raised" without traceback
+ # Worker may have crashed before ack
- @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}"
+ # 3. Verify task version probe
+ celery -A myproject inspect registered # Compare task versions across workers
```
- ### Task Options
+ **Fix:**
+ - Ensure `broker_url` and `result_backend` point to same broker instance
+ - Set `CELERY_TASK_ALWAYS_EAGER = False` (debug mode default)
+ - Check firewall/network between workers and broker
+ ### Symptom: Stale Workers Running Old Code
+
+ **How to confirm:**
```python
- from celery import shared_task
+ # In task, probe version
+ @shared_task
+ def version_probe():
+ import myapp.tasks
+ return myapp.tasks.__version__ # Or check git commit hash
- @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)
+ # Compare across workers
+ celery -A myproject inspect active
```
- ### Bound Tasks
+ **Fix:**
+ - Restart workers after code deploy: `systemctl restart celery`
+ - Use `--max-tasks-per-child=1000` to force periodic restarts
+ - Deploy with zero-downtime: start new workers, drain old workers
- ```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)
+ ### Symptom: Worker Killed Mid-Task (Data Corruption)
+
+ **How to confirm:**
+ ```bash
+ # Check for OOM killer in system logs
+ dmesg | grep -i "out of memory"
+ grep "Out of memory" /var/log/syslog
+
+ # Check worker process exit codes
+ journalctl -u celery | grep "killed"
```
- ### Task Inheritance
+ **Fix:**
+ - Set `acks_late=True` + idempotent tasks (task re-runs on crash)
+ - Increase worker memory limits
+ - Use `CELERY_WORKER_MAX_MEMORY_PER_CHILD = 400000` (KB)
+ ### Symptom: Time-Limit Exceptions
+
```python
- from celery import Task
+ # Hard time limit: Worker kills task immediately
+ @shared_task(time_limit=300) # 5 minutes hard limit
+ def long_task():
+ pass
- 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
+ # Soft time limit: Task receives SoftTimeLimitExceeded exception
+ @shared_task(soft_time_limit=240) # 4 minutes soft limit
+ def graceful_task():
+ try:
+ return do_work()
+ except SoftTimeLimitExceeded:
+ # Cleanup before forced termination
+ cleanup()
+ raise
+ ```
- @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()
+ **Detection:**
+ ```bash
+ # Check worker logs for "Task raised SoftTimeLimitExceeded"
+ grep "SoftTimeLimitExceeded" /var/log/celery/worker.log
```
- ### Task Signals
+ ### Symptom: Results Never Appear
+ **How to confirm:**
```python
- from celery import signals
+ # 1. Check result backend is configured
+ print(app.conf.result_backend) # Must not be None
- @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...")
+ # 2. Check backend matches between sender and receiver
+ # If task sent with backend=redis://... but result.get() uses different backend
- @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}")
+ # 3. Check result expiration
+ print(app.conf.result_expires) # Default 1 day
+ ```
- @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}")
+ **Fix:**
+ - Set `CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'`
+ - Don't set `ignore_result=True` unless you don't need results
+ - Increase `result_expires` for long-running workflows
- @signals.task_retry.connect
- def task_retry_handler(sender=None, reason=None, **kwargs):
- """Called on task retry."""
- print(f"Task retrying: {reason}")
+ ### Symptom: Duplicate Task Execution
+
+ **How to confirm:**
+ ```bash
+ # Check acks_late setting
+ celery -A myproject inspect config | grep ACKS_LATE
+
+ # Check for worker restarts during task execution
+ journalctl -u celery | grep -E "(restarting|killed|OOM)"
```
+ **Fix:**
+ - Make tasks idempotent (see retry section)
+ - Use database unique constraints
+ - Implement deduplication tokens
+
## Best Practices
- ### 1. Idempotent Tasks
+ ### Idempotent Tasks
```python
@shared_task(bind=True)
def process_payment(self, payment_id):
- """Idempotent payment processing."""
+ """Idempotent: Safe to run multiple times."""
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():
+ payment.refresh_from_db()
+ if payment.status == 'completed':
+ return {'status': 'already_processed'}
+
result = payment.charge()
- payment.mark_completed()
+ payment.status = 'completed'
+ payment.save()
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
+ ### Task Granularity
```python
- # Bad: One monolithic task
+ # Bad: 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
+ # Good: Composable tasks
@shared_task
def validate_order(order_id):
- order = Order.objects.get(id=order_id)
- order.validate()
+ Order.objects.get(id=order_id).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
- ```
-
## Deep Dives
- For detailed reference on specific topics, load these on-demand:
+ Load these on demand for detailed reference:
- - **Calling Workflows** (`references/calling-workflows.md`) — Task invocation patterns: `apply_async`, signatures, chains, groups, chords, chunks, and task states
+ - **Calling Workflows** (`references/calling-workflows.md`) — `apply_async` options, 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
+ - **Testing & Performance** (`references/testing-performance.md`) — pytest testing, worker configuration, task optimization, memory management
## 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
+