Django Celery Expert

Django Celery Expert

Instructions

Step 1: Classify the Request

Identify the task category from the request:

• Django integration — transaction safety, ORM patterns, testing, request correlation → read references/django-integration.md
• Task design — new tasks, calling patterns, chains/groups/chords, idempotency → read references/task-design-patterns.md
• Configuration — broker setup, result backend, worker settings, queue routing → read references/configuration-guide.md
• Error handling — retries, backoff, dead letter queues, timeouts → read references/error-handling.md
• Periodic tasks — Celery Beat, crontab schedules, dynamic schedules, timezone handling → read references/periodic-tasks.md
• Monitoring — Flower, Prometheus, logging, debugging stuck tasks → read references/monitoring-observability.md
• Production deployment — scaling, supervision, containers, health checks → read references/production-deployment.md

If the request spans multiple categories, read all relevant reference files before continuing.

Step 2: Read the Reference File(s)

Read each reference file identified in Step 1. Do not proceed to implementation without reading the relevant reference.

Step 3: Implement

Apply the patterns from the reference file. Before presenting the solution, verify:

• Task arguments are serializable (pass IDs, not model instances)
• Tasks with retries enabled are idempotent
• Errors are logged with context
• Long-running tasks have timeouts configured

Examples

Basic Background Task

Request: "Send welcome emails in the background after user registration"

# tasks.py
from celery import shared_task
from django.core.mail import send_mail

@shared_task(bind=True, max_retries=3)
def send_welcome_email(self, user_id):
    from users.models import User

    try:
        user = User.objects.get(id=user_id)
        send_mail(
            subject="Welcome!",
            message=f"Hi {user.name}, welcome to our platform!",
            from_email="noreply@example.com",
            recipient_list=[user.email],
        )
    except User.DoesNotExist:
        pass
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))


# views.py — queue only after the transaction commits
from django.db import transaction

def register(request):
    user = User.objects.create(...)
    transaction.on_commit(lambda: send_welcome_email.delay(user.id))
    return redirect("dashboard")

Task with Progress Tracking

Request: "Process a large CSV import with progress updates"

@shared_task(bind=True)
def import_csv(self, file_path, total_rows):
    from myapp.models import Record

    with open(file_path) as f:
        reader = csv.DictReader(f)
        for i, row in enumerate(reader):
            Record.objects.create(**row)
            if i % 100 == 0:
                self.update_state(
                    state="PROGRESS",
                    meta={"current": i, "total": total_rows},
                )

    return {"status": "complete", "processed": total_rows}


# Poll progress
result = import_csv.AsyncResult(task_id)
if result.state == "PROGRESS":
    progress = result.info.get("current", 0) / result.info.get("total", 1)

Workflow with Chains

Request: "Process an order: validate inventory, charge payment, then send confirmation"

from celery import chain

@shared_task
def validate_inventory(order_id):
    order = Order.objects.get(id=order_id)
    if not order.items_in_stock():
        raise ValueError("Items out of stock")
    return order_id

@shared_task
def charge_payment(order_id):
    order = Order.objects.get(id=order_id)
    order.charge()
    return order_id

@shared_task
def send_confirmation(order_id):
    Order.objects.get(id=order_id).send_confirmation_email()

def process_order(order_id):
    chain(
        validate_inventory.s(order_id),
        charge_payment.s(),
        send_confirmation.s(),
    ).delay()

Configuration Guide

Django-Celery Integration Setup

Project Structure

myproject/
├── myproject/
│   ├── __init__.py
│   ├── celery.py      # Celery app configuration
│   ├── settings.py
│   └── urls.py
├── myapp/
│   ├── __init__.py
│   ├── models.py
│   └── tasks.py       # App-specific tasks
└── manage.py

Celery Application Setup

# myproject/celery.py
import os
from celery import Celery

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 all installed apps
app.autodiscover_tasks()

@app.task(bind=True, ignore_result=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

# myproject/__init__.py
from .celery import app as celery_app

__all__ = ('celery_app',)

Django Settings

# settings.py

# Broker Configuration (choose one)
# Redis
CELERY_BROKER_URL = 'redis://localhost:6379/0'

# RabbitMQ
CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672//'

# Result Backend (optional, needed if you track results)
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
# Or use Django database
CELERY_RESULT_BACKEND = 'django-db'

# Serialization
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']

# Timezone
CELERY_TIMEZONE = 'UTC'
CELERY_ENABLE_UTC = True

# Task settings
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60  # 30 minutes hard limit
CELERY_TASK_SOFT_TIME_LIMIT = 25 * 60  # 25 minutes soft limit

Broker Configuration

Redis Broker

# Basic Redis
CELERY_BROKER_URL = 'redis://localhost:6379/0'

# Redis with password
CELERY_BROKER_URL = 'redis://:password@localhost:6379/0'

# Redis Sentinel for HA
CELERY_BROKER_URL = 'sentinel://localhost:26379'
CELERY_BROKER_TRANSPORT_OPTIONS = {
    'master_name': 'mymaster',
    'sentinel_kwargs': {'password': 'sentinel_password'},
}

# Redis Cluster
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_BROKER_TRANSPORT_OPTIONS = {
    'cluster': True,
}

# Connection pool settings
CELERY_BROKER_POOL_LIMIT = 10
CELERY_BROKER_CONNECTION_TIMEOUT = 10
CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP = True

RabbitMQ Broker

# Basic RabbitMQ
CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672//'

# With virtual host
CELERY_BROKER_URL = 'amqp://user:password@localhost:5672/myvhost'

# RabbitMQ with SSL
CELERY_BROKER_URL = 'amqp://user:password@localhost:5671//'
CELERY_BROKER_USE_SSL = {
    'keyfile': '/path/to/key.pem',
    'certfile': '/path/to/cert.pem',
    'ca_certs': '/path/to/ca.pem',
    'cert_reqs': ssl.CERT_REQUIRED,
}

# Connection pool
CELERY_BROKER_POOL_LIMIT = 10
CELERY_BROKER_HEARTBEAT = 10

# Confirm message delivery (recommended for reliability)
CELERY_BROKER_TRANSPORT_OPTIONS = {
    'confirm_publish': True,
}

Broker Transport Options

Critical settings for message delivery reliability.

Redis Visibility Timeout

When using Redis, set visibility_timeout to control how long a task remains invisible after being picked up. If a worker crashes before acknowledging, the task becomes visible again after this timeout.

# Redis visibility timeout (default: 1 hour)
CELERY_BROKER_TRANSPORT_OPTIONS = {
    'visibility_timeout': 3600,  # 1 hour in seconds
}

Important: Set visibility timeout longer than your longest task. If a task takes longer than the timeout, it will be redelivered to another worker while still running.

RabbitMQ Publisher Confirms

Enable publisher confirms to ensure messages are actually delivered to the broker.

# RabbitMQ: confirm message delivery
CELERY_BROKER_TRANSPORT_OPTIONS = {
    'confirm_publish': True,
}

SQS Configuration

# AWS SQS
CELERY_BROKER_URL = 'sqs://aws_access_key:aws_secret_key@'
CELERY_BROKER_TRANSPORT_OPTIONS = {
    'region': 'us-east-1',
    'visibility_timeout': 3600,
    'polling_interval': 1,
    'queue_name_prefix': 'celery-',
}

Separating Redis Instances

Critical: Use separate Redis instances for different purposes to prevent cascade failures.

# BAD: Single Redis for everything
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CACHES = {'default': {'BACKEND': '...', 'LOCATION': 'redis://localhost:6379/0'}}

# GOOD: Separate Redis instances/databases
CELERY_BROKER_URL = 'redis://localhost:6379/0'      # Task queue
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'  # Results
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': 'redis://localhost:6379/2',     # Cache
    }
}

# BEST: Separate Redis servers in production
CELERY_BROKER_URL = 'redis://redis-queue:6379/0'
CELERY_RESULT_BACKEND = 'redis://redis-results:6379/0'
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': 'redis://redis-cache:6379/0',
    }
}

Why separate?

• Cache eviction won't affect task queue
• Result backend memory pressure won't block new tasks
• Easier to scale each component independently
• Isolates failures

Result Backend Configuration

Redis Result Backend

CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'

# Result expiration (default: 1 day)
CELERY_RESULT_EXPIRES = 60 * 60 * 24  # 24 hours

# Extended result info
CELERY_RESULT_EXTENDED = True

Django Database Result Backend

# Install: pip install django-celery-results
INSTALLED_APPS = [
    ...
    'django_celery_results',
]

CELERY_RESULT_BACKEND = 'django-db'

# Run migrations
# python manage.py migrate django_celery_results

# Optional: Cache results in addition to database
CELERY_CACHE_BACKEND = 'django-cache'

Disable Results (Fire-and-Forget)

# Global: disable result tracking
CELERY_TASK_IGNORE_RESULT = True

# Per-task: disable result
@shared_task(ignore_result=True)
def fire_and_forget_task(data):
    pass

Worker Configuration

Concurrency Settings

# Number of worker processes/threads
CELERY_WORKER_CONCURRENCY = 4  # Default: CPU count

# Prefetch multiplier (tasks to prefetch per worker)
CELERY_WORKER_PREFETCH_MULTIPLIER = 1  # For fair distribution
# Use 4 (default) for throughput, 1 for latency-sensitive tasks

# Max tasks per worker before restart (memory leak prevention)
CELERY_WORKER_MAX_TASKS_PER_CHILD = 1000

Task Execution Settings

# Task time limits
CELERY_TASK_TIME_LIMIT = 300  # Hard kill after 5 min
CELERY_TASK_SOFT_TIME_LIMIT = 240  # Raise exception after 4 min

# Task acknowledgment
CELERY_TASK_ACKS_LATE = True  # Ack after task completes (for reliability)
CELERY_TASK_REJECT_ON_WORKER_LOST = True  # Requeue if worker dies

# Task result compression
CELERY_RESULT_COMPRESSION = 'gzip'

Task Routing and Queues

Basic Queue Configuration

# Define queues
from kombu import Queue

CELERY_TASK_QUEUES = (
    Queue('default', routing_key='default'),
    Queue('high-priority', routing_key='high'),
    Queue('low-priority', routing_key='low'),
    Queue('emails', routing_key='emails'),
)

CELERY_TASK_DEFAULT_QUEUE = 'default'

Task Routing

# Route tasks to specific queues
CELERY_TASK_ROUTES = {
    # By task name
    'myapp.tasks.send_email': {'queue': 'emails'},
    'myapp.tasks.critical_task': {'queue': 'high-priority'},

    # By pattern
    'myapp.tasks.report_*': {'queue': 'low-priority'},

    # By module
    'myapp.email_tasks.*': {'queue': 'emails'},
}

# Or use a function for dynamic routing
def route_task(name, args, kwargs, options, task=None, **kw):
    if 'urgent' in kwargs.get('priority', ''):
        return {'queue': 'high-priority'}
    return {'queue': 'default'}

CELERY_TASK_ROUTES = (route_task,)

Running Workers for Specific Queues

# Process only high-priority queue
celery -A myproject worker -Q high-priority

# Process multiple queues with priority
celery -A myproject worker -Q high-priority,default,low-priority

# Dedicated email worker
celery -A myproject worker -Q emails -c 2 -n email-worker@%h

Serialization

Serializer Options

# JSON (default, safe, recommended)
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']

# Pickle (supports Python objects, security risk!)
CELERY_TASK_SERIALIZER = 'pickle'
CELERY_RESULT_SERIALIZER = 'pickle'
CELERY_ACCEPT_CONTENT = ['pickle', 'json']

# MessagePack (binary, faster than JSON)
# pip install msgpack
CELERY_TASK_SERIALIZER = 'msgpack'
CELERY_RESULT_SERIALIZER = 'msgpack'
CELERY_ACCEPT_CONTENT = ['msgpack', 'json']

Per-Task Serializer

@shared_task(serializer='pickle')
def task_with_complex_objects(data):
    # Can receive Python objects
    pass

Security Settings

Secure Serialization

# Only accept JSON (never use pickle in production with untrusted data)
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

# Content signing (for pickle if absolutely needed)
CELERY_TASK_SERIALIZER = 'auth'
CELERY_SECURITY_KEY = '/path/to/private.key'
CELERY_SECURITY_CERTIFICATE = '/path/to/certificate.pem'
CELERY_SECURITY_CERT_STORE = '/path/to/certs/'

Rate Limiting

# Global rate limit (tasks/second)
CELERY_TASK_DEFAULT_RATE_LIMIT = '100/s'

# Per-task rate limit
@shared_task(rate_limit='10/m')  # 10 per minute
def rate_limited_task():
    pass

Environment-Specific Configuration

Development Settings

# settings/development.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'

# Eager mode: execute tasks synchronously (for debugging)
CELERY_TASK_ALWAYS_EAGER = True
CELERY_TASK_EAGER_PROPAGATES = True

Production Settings

# settings/production.py
import os

CELERY_BROKER_URL = os.environ['CELERY_BROKER_URL']
CELERY_RESULT_BACKEND = os.environ['CELERY_RESULT_BACKEND']

# Never use eager mode in production
CELERY_TASK_ALWAYS_EAGER = False

# Connection resilience
CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP = True
CELERY_BROKER_CONNECTION_MAX_RETRIES = 10

# Task settings
CELERY_TASK_ACKS_LATE = True
CELERY_TASK_REJECT_ON_WORKER_LOST = True
CELERY_WORKER_PREFETCH_MULTIPLIER = 1

# Result expiration
CELERY_RESULT_EXPIRES = 60 * 60 * 24  # 24 hours

# Memory management
CELERY_WORKER_MAX_TASKS_PER_CHILD = 1000

Testing Settings

# settings/testing.py
CELERY_TASK_ALWAYS_EAGER = True
CELERY_TASK_EAGER_PROPAGATES = True
CELERY_BROKER_URL = 'memory://'
CELERY_RESULT_BACKEND = 'cache+memory://'

Complete Configuration Example

# settings.py - Complete Celery configuration

from kombu import Queue
import os

# Broker
CELERY_BROKER_URL = os.environ.get('CELERY_BROKER_URL', 'redis://localhost:6379/0')
CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP = True

# Result backend
CELERY_RESULT_BACKEND = os.environ.get('CELERY_RESULT_BACKEND', 'django-db')
CELERY_RESULT_EXPIRES = 60 * 60 * 24  # 24 hours

# Serialization
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']

# Time and timezone
CELERY_TIMEZONE = 'UTC'
CELERY_ENABLE_UTC = True

# Task execution
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60  # 30 min hard limit
CELERY_TASK_SOFT_TIME_LIMIT = 25 * 60  # 25 min soft limit
CELERY_TASK_ACKS_LATE = True
CELERY_TASK_REJECT_ON_WORKER_LOST = True

# Worker
CELERY_WORKER_PREFETCH_MULTIPLIER = 1
CELERY_WORKER_MAX_TASKS_PER_CHILD = 1000

# Queues
CELERY_TASK_QUEUES = (
    Queue('default'),
    Queue('high-priority'),
    Queue('low-priority'),
)
CELERY_TASK_DEFAULT_QUEUE = 'default'

# Routing
CELERY_TASK_ROUTES = {
    'myapp.tasks.urgent_*': {'queue': 'high-priority'},
    'myapp.tasks.report_*': {'queue': 'low-priority'},
}

# Beat scheduler (for periodic tasks)
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

Django Integration Patterns

Critical patterns for reliable Django and Celery integration.

Sources:

• Celery in the Wild: Tips and Tricks to Run Async Tasks in the Real World - Vinta Software
• A Guide on Django Celery Tasks That Actually Work - Vinta Software

Critical: Use transaction.on_commit()

Always use transaction.on_commit() when queuing tasks after database writes. This ensures the database transaction commits before the task is queued, preventing race conditions where the task runs before the data is available.

from django.db import transaction

class SignUpView(CreateView):
    def form_valid(self, form):
        response = super().form_valid(form)
        user_pk = self.object.pk

        # GOOD: Task queued only after transaction commits
        transaction.on_commit(
            lambda: send_activation_email.delay(user_pk)
        )
        return response

# BAD: Task might run before user is committed to DB
def bad_signup(request):
    user = User.objects.create(...)
    send_activation_email.delay(user.pk)  # Race condition!
    return redirect('home')

With Atomic Blocks

from django.db import transaction

def process_order(order_id):
    with transaction.atomic():
        order = Order.objects.select_for_update().get(id=order_id)
        order.status = 'processing'
        order.save()

        # Queue task only after atomic block commits
        transaction.on_commit(
            lambda: notify_warehouse.delay(order.pk)
        )

Nested Transactions

from django.db import transaction

def complex_operation():
    with transaction.atomic():
        create_parent_record()

        with transaction.atomic():
            create_child_record()
            # This on_commit waits for OUTER transaction
            transaction.on_commit(lambda: child_task.delay())

        # This also waits for outer transaction
        transaction.on_commit(lambda: parent_task.delay())

Database as Source of Truth

Don't rely solely on Celery for guaranteed delivery. Use the database as the source of truth and implement recovery tasks.

Pattern: Flag-Based Recovery

# models.py
class User(models.Model):
    email = models.EmailField()
    is_activation_email_sent = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)

# tasks.py
@shared_task(bind=True, max_retries=3)
def send_activation_email(self, user_pk):
    try:
        user = User.objects.get(pk=user_pk)
        if user.is_activation_email_sent:
            return  # Already sent, idempotent

        send_email(user.email, 'Activate your account')

        # Mark as sent atomically
        User.objects.filter(pk=user_pk).update(is_activation_email_sent=True)

    except User.DoesNotExist:
        pass  # User deleted, skip
    except EmailException as exc:
        raise self.retry(exc=exc)

# Recovery task - runs periodically via Celery Beat
@shared_task
def recover_unsent_activation_emails():
    """Catch any missed activation emails."""
    cutoff = timezone.now() - timedelta(hours=24)
    unsent = User.objects.filter(
        is_activation_email_sent=False,
        created_at__lt=timezone.now() - timedelta(minutes=5),  # Grace period
        created_at__gt=cutoff,  # Don't process very old records
    )
    for user_pk in unsent.values_list('pk', flat=True):
        send_activation_email.delay(user_pk)

Celery Beat Schedule for Recovery

CELERY_BEAT_SCHEDULE = {
    'recover-activation-emails': {
        'task': 'users.tasks.recover_unsent_activation_emails',
        'schedule': crontab(minute='*/15'),  # Every 15 minutes
    },
}

Atomicity: External Calls Before DB Writes

Execute external API calls before modifying the database. This prevents inconsistent states where the DB is updated but the external call fails.

# GOOD: External call first, then DB update
@shared_task(bind=True, max_retries=3)
def sync_user_to_crm(self, user_id):
    user = User.objects.get(id=user_id)

    try:
        # External call FIRST
        crm_response = crm_api.create_contact(
            email=user.email,
            name=user.name,
        )

        # DB update AFTER external call succeeds
        user.crm_id = crm_response['id']
        user.crm_synced_at = timezone.now()
        user.save()

    except CRMAPIError as exc:
        raise self.retry(exc=exc)

# BAD: DB updated before external call
@shared_task
def bad_sync_user_to_crm(user_id):
    user = User.objects.get(id=user_id)
    user.crm_sync_started = True
    user.save()  # DB updated

    crm_api.create_contact(...)  # This might fail!
    # Now DB is in inconsistent state

Request-Task Correlation with django-guid

Use django-guid to trace requests through to Celery tasks for debugging.

pip install django-guid celery-guid

# settings.py
INSTALLED_APPS = [
    ...
    'django_guid',
]

MIDDLEWARE = [
    'django_guid.middleware.guid_middleware',
    ...
]

DJANGO_GUID = {
    'GUID_HEADER_NAME': 'X-Correlation-ID',
    'VALIDATE_GUID': False,
    'RETURN_HEADER': True,
    'INTEGRATIONS': [
        'django_guid.integrations.celery.CeleryIntegration',
    ],
}

# celery.py
from celery import Celery
from django_guid.integrations.celery import CeleryIntegration

app = Celery('myproject')

# Logging format includes correlation ID
LOGGING = {
    'formatters': {
        'correlation': {
            'format': '[%(correlation_id)s] %(levelname)s %(name)s: %(message)s',
        },
    },
}

Manual Correlation Without django-guid

import uuid
from celery import shared_task

@shared_task(bind=True)
def task_with_correlation(self, data):
    correlation_id = self.request.headers.get('correlation_id', 'unknown')
    logger.info(f"[{correlation_id}] Processing task")
    # ... task logic

# Pass correlation ID when calling
def my_view(request):
    correlation_id = request.headers.get('X-Correlation-ID', str(uuid.uuid4()))
    task_with_correlation.apply_async(
        args=[data],
        headers={'correlation_id': correlation_id}
    )

Task-Specific Logging

Use Celery's task logger for proper task context in logs.

from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@shared_task(bind=True)
def process_order(self, order_id):
    logger.info(f'Starting order processing', extra={
        'order_id': order_id,
        'task_id': self.request.id,
    })

    try:
        order = Order.objects.get(id=order_id)
        # ... process order
        logger.info(f'Order processed successfully')

    except Order.DoesNotExist:
        logger.warning(f'Order not found: {order_id}')

    except Exception as exc:
        logger.exception(f'Order processing failed')
        raise

Idempotent ORM Operations

Use Django ORM methods that are naturally idempotent for safe retries.

@shared_task(bind=True, max_retries=5)
def update_user_stats(self, user_id, points_earned):
    try:
        # GOOD: Idempotent with get_or_create
        stats, created = UserStats.objects.get_or_create(
            user_id=user_id,
            defaults={'points': points_earned}
        )

        if not created:
            # GOOD: Idempotent with update_or_create
            UserStats.objects.update_or_create(
                user_id=user_id,
                defaults={'points': F('points') + points_earned}
            )

    except Exception as exc:
        raise self.retry(exc=exc)

# For truly idempotent operations, use unique constraints
@shared_task
def record_event(event_id, user_id, event_type):
    # Idempotency via unique constraint
    Event.objects.get_or_create(
        idempotency_key=event_id,
        defaults={
            'user_id': user_id,
            'event_type': event_type,
        }
    )

Handling Model Changes in Tasks

Tasks may run with stale model data. Always fetch fresh data.

# BAD: Passing model state that might be stale
@shared_task
def bad_task(user_email, user_name):
    send_email(user_email, f"Hello {user_name}")
    # What if user updated their email before task ran?

# GOOD: Fetch fresh data by ID
@shared_task
def good_task(user_id):
    user = User.objects.get(id=user_id)  # Fresh from DB
    send_email(user.email, f"Hello {user.name}")

Testing Django Celery Tasks

Unit Testing with Eager Mode

# settings/test.py
CELERY_TASK_ALWAYS_EAGER = True
CELERY_TASK_EAGER_PROPAGATES = True

# test_tasks.py
from django.test import TestCase, override_settings

@override_settings(CELERY_TASK_ALWAYS_EAGER=True)
class TaskTestCase(TestCase):
    def test_send_email_task(self):
        user = User.objects.create(email='test@example.com')
        result = send_activation_email.delay(user.pk)

        # Task runs synchronously in tests
        self.assertTrue(result.successful())
        user.refresh_from_db()
        self.assertTrue(user.is_activation_email_sent)

Testing with Mocked Tasks

from unittest.mock import patch

class ViewTestCase(TestCase):
    @patch('myapp.tasks.send_activation_email.delay')
    def test_signup_queues_email(self, mock_task):
        response = self.client.post('/signup/', {
            'email': 'test@example.com',
            'password': 'securepass123',
        })

        self.assertEqual(response.status_code, 302)
        mock_task.assert_called_once()

Testing transaction.on_commit

from django.test import TestCase, TransactionTestCase

class OnCommitTestCase(TransactionTestCase):
    """Use TransactionTestCase for on_commit testing."""

    @patch('myapp.tasks.send_email.delay')
    def test_email_sent_after_commit(self, mock_task):
        # Task is queued after transaction commits
        User.objects.create(email='test@example.com')

        # In TransactionTestCase, on_commit callbacks execute
        mock_task.assert_called_once()

Error Handling and Retry Strategies

Basic Retry Configuration

Simple Retry

from celery import shared_task

@shared_task(bind=True, max_retries=3)
def task_with_retry(self, data):
    try:
        process(data)
    except TemporaryError as exc:
        # Retry with default delay (3 minutes)
        raise self.retry(exc=exc)

Custom Retry Delay

@shared_task(bind=True, max_retries=5)
def task_with_custom_retry(self, data):
    try:
        process(data)
    except TemporaryError as exc:
        # Retry after 60 seconds
        raise self.retry(exc=exc, countdown=60)

Exponential Backoff

@shared_task(bind=True, max_retries=5)
def task_with_backoff(self, data):
    try:
        process(data)
    except TemporaryError as exc:
        # Exponential backoff: 1min, 2min, 4min, 8min, 16min
        countdown = 60 * (2 ** self.request.retries)
        raise self.retry(exc=exc, countdown=countdown)

Retry with Jitter

import random

@shared_task(bind=True, max_retries=5)
def task_with_jitter(self, data):
    try:
        process(data)
    except TemporaryError as exc:
        # Add randomness to prevent thundering herd
        base_delay = 60 * (2 ** self.request.retries)
        jitter = random.uniform(0, base_delay * 0.1)
        raise self.retry(exc=exc, countdown=base_delay + jitter)

Auto-Retry Decorator

Using autoretry_for

@shared_task(
    bind=True,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    retry_backoff_max=600,  # Max 10 minutes between retries
    retry_jitter=True,
    max_retries=5,
)
def auto_retry_task(self, data):
    # Will automatically retry on ConnectionError or TimeoutError
    response = call_external_api(data)
    return response

Multiple Exception Types

@shared_task(
    autoretry_for=(
        ConnectionError,
        TimeoutError,
        requests.exceptions.RequestException,
    ),
    retry_backoff=60,  # Start with 60 second delay
    retry_backoff_max=3600,  # Max 1 hour
    max_retries=10,
)
def robust_api_call(url):
    return requests.get(url, timeout=30).json()

Exception Handling Patterns

Handle Specific Exceptions Differently

from celery.exceptions import MaxRetriesExceededError

@shared_task(bind=True, max_retries=3)
def task_with_exception_handling(self, order_id):
    try:
        order = Order.objects.get(id=order_id)
        order.process()

    except Order.DoesNotExist:
        # Permanent failure - don't retry
        logger.error(f"Order {order_id} not found")
        return None

    except PaymentGatewayError as exc:
        # Temporary - retry with backoff
        logger.warning(f"Payment gateway error: {exc}")
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))

    except InsufficientFundsError:
        # Business logic failure - don't retry
        order.mark_payment_failed()
        return {'status': 'failed', 'reason': 'insufficient_funds'}

    except MaxRetriesExceededError:
        # All retries exhausted
        order.mark_requires_manual_review()
        notify_admin(f"Order {order_id} requires manual review")
        raise

Cleanup on Failure

@shared_task(bind=True, max_retries=3)
def task_with_cleanup(self, resource_id):
    resource = None
    try:
        resource = acquire_resource(resource_id)
        process(resource)
    except Exception as exc:
        if self.request.retries >= self.max_retries:
            # Final failure - cleanup
            if resource:
                cleanup_resource(resource)
            mark_as_failed(resource_id)
        raise self.retry(exc=exc, countdown=60)
    finally:
        if resource:
            release_resource(resource)

Timeout Handling

Soft Time Limit

from celery.exceptions import SoftTimeLimitExceeded

@shared_task(
    bind=True,
    soft_time_limit=300,  # Raise exception after 5 min
    time_limit=360,       # Kill process after 6 min
)
def task_with_timeout(self, large_dataset):
    try:
        for item in large_dataset:
            process_item(item)
    except SoftTimeLimitExceeded:
        # Graceful cleanup
        logger.warning("Task timeout approaching, saving progress")
        save_progress()
        # Optionally retry with remaining items
        raise self.retry(countdown=60)

Per-Execution Timeout

@shared_task(bind=True)
def dynamic_timeout_task(self, data, timeout=300):
    # Apply timeout dynamically
    self.request.timelimit = (timeout, timeout + 60)
    process(data)

Dead Letter Queue Pattern

Manual Dead Letter Queue

from kombu import Queue

CELERY_TASK_QUEUES = (
    Queue('default'),
    Queue('dead_letter'),
)

@shared_task(bind=True, max_retries=3)
def task_with_dlq(self, data):
    try:
        process(data)
    except Exception as exc:
        if self.request.retries >= self.max_retries:
            # Send to dead letter queue for manual processing
            handle_dead_letter.delay(
                task_name=self.name,
                args=self.request.args,
                kwargs=self.request.kwargs,
                exception=str(exc),
                traceback=traceback.format_exc(),
            )
            return
        raise self.retry(exc=exc, countdown=60)


@shared_task(queue='dead_letter')
def handle_dead_letter(task_name, args, kwargs, exception, traceback):
    # Store for manual review
    DeadLetterMessage.objects.create(
        task_name=task_name,
        args=args,
        kwargs=kwargs,
        exception=exception,
        traceback=traceback,
    )
    notify_admin(f"Dead letter: {task_name}")

RabbitMQ Dead Letter Exchange

from kombu import Exchange, Queue

# Configure dead letter exchange in RabbitMQ
dlx = Exchange('dlx', type='direct')
dead_letter_queue = Queue(
    'dead_letters',
    exchange=dlx,
    routing_key='dead_letter',
)

main_queue = Queue(
    'default',
    queue_arguments={
        'x-dead-letter-exchange': 'dlx',
        'x-dead-letter-routing-key': 'dead_letter',
    },
)

CELERY_TASK_QUEUES = (main_queue, dead_letter_queue)

Error Callbacks

Using on_failure

from celery import Task

class TaskWithErrorCallback(Task):
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        logger.error(
            f'Task {self.name}[{task_id}] failed: {exc}',
            exc_info=einfo.exc_info,
        )
        # Send notification
        send_alert(
            title=f"Task Failed: {self.name}",
            message=str(exc),
            task_id=task_id,
        )


@shared_task(base=TaskWithErrorCallback, bind=True)
def task_with_error_callback(self, data):
    process(data)

Link Error Callback

@shared_task
def error_handler(request, exc, traceback):
    logger.error(
        f'Task {request.id} raised exception: {exc}',
        extra={
            'task_id': request.id,
            'task_name': request.task,
            'args': request.args,
            'kwargs': request.kwargs,
        }
    )

# Attach error callback when calling
my_task.apply_async(
    args=[data],
    link_error=error_handler.s(),
)

Idempotent Error Handling

Safe Retry Pattern

@shared_task(bind=True, max_retries=5)
def idempotent_payment_task(self, order_id, idempotency_key):
    # Check if already processed
    if PaymentRecord.objects.filter(idempotency_key=idempotency_key).exists():
        logger.info(f"Payment {idempotency_key} already processed")
        return {'status': 'already_processed'}

    try:
        result = process_payment(order_id)

        # Record successful payment
        PaymentRecord.objects.create(
            idempotency_key=idempotency_key,
            order_id=order_id,
            result=result,
        )
        return result

    except PaymentGatewayError as exc:
        # Safe to retry - payment won't be duplicated
        raise self.retry(exc=exc, countdown=60)

Transactional Safety

from django.db import transaction

@shared_task(bind=True, max_retries=3)
def transactional_task(self, order_id):
    try:
        with transaction.atomic():
            order = Order.objects.select_for_update().get(id=order_id)

            if order.status == 'processed':
                return {'status': 'already_processed'}

            # All-or-nothing processing
            process_order(order)
            update_inventory(order)
            order.status = 'processed'
            order.save()

    except Order.DoesNotExist:
        return None
    except Exception as exc:
        # Transaction rolled back, safe to retry
        raise self.retry(exc=exc, countdown=60)

Monitoring Failed Tasks

Logging Configuration

# celery.py
import logging

logger = logging.getLogger('celery.task')

@signals.task_failure.connect
def log_task_failure(sender, task_id, exception, args, kwargs, traceback, einfo, **kw):
    logger.error(
        f'Task {sender.name}[{task_id}] failed',
        exc_info=einfo.exc_info,
        extra={
            'task_name': sender.name,
            'task_id': task_id,
            'args': args,
            'kwargs': kwargs,
        }
    )

@signals.task_retry.connect
def log_task_retry(sender, request, reason, einfo, **kw):
    logger.warning(
        f'Task {sender.name}[{request.id}] retrying: {reason}',
        extra={
            'task_name': sender.name,
            'task_id': request.id,
            'retry_count': request.retries,
        }
    )

Storing Failed Tasks

# models.py
class FailedTask(models.Model):
    task_id = models.CharField(max_length=255, unique=True)
    task_name = models.CharField(max_length=255)
    args = models.JSONField()
    kwargs = models.JSONField()
    exception = models.TextField()
    traceback = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)
    retried = models.BooleanField(default=False)

# signals
@signals.task_failure.connect
def store_failed_task(sender, task_id, exception, args, kwargs, traceback, einfo, **kw):
    FailedTask.objects.update_or_create(
        task_id=task_id,
        defaults={
            'task_name': sender.name,
            'args': list(args),
            'kwargs': dict(kwargs),
            'exception': str(exception),
            'traceback': einfo.traceback if einfo else '',
        }
    )

Retry Best Practices Summary

Configuration Recommendations

# Production retry configuration
@shared_task(
    bind=True,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,        # Exponential backoff
    retry_backoff_max=600,     # Max 10 min between retries
    retry_jitter=True,         # Add randomness
    max_retries=5,             # Limit total attempts
    acks_late=True,            # Ack after completion
    reject_on_worker_lost=True # Requeue if worker dies
)
def production_task(self, data):
    pass

Monitoring and Observability

Flower - Real-Time Monitor

Installation and Setup

pip install flower

# Basic usage
celery -A myproject flower

# With options
celery -A myproject flower \
    --port=5555 \
    --broker=redis://localhost:6379/0 \
    --basic_auth=admin:password

# Persistent storage
celery -A myproject flower --persistent=True --db=flower.db

Flower Configuration

# flower_config.py
broker_api = 'redis://localhost:6379/0'
port = 5555
basic_auth = ['admin:secure_password']
persistent = True
db = '/var/lib/flower/flower.db'
max_tasks = 10000
purge_offline_workers = 300  # Remove offline workers after 5 min

celery -A myproject flower --conf=flower_config.py

Flower in Production

# docker-compose.yml
services:
  flower:
    image: mher/flower
    command: celery flower --broker=redis://redis:6379/0
    ports:
      - "5555:5555"
    environment:
      - FLOWER_BASIC_AUTH=admin:password
    depends_on:
      - redis
      - worker

Flower API

import requests

# List workers
response = requests.get('http://localhost:5555/api/workers')
workers = response.json()

# Get task info
response = requests.get(f'http://localhost:5555/api/task/info/{task_id}')
task_info = response.json()

# Revoke task
requests.post(f'http://localhost:5555/api/task/revoke/{task_id}')

# Queue length
response = requests.get('http://localhost:5555/api/queues/length')
queue_lengths = response.json()

Logging Best Practices

Task Logging Configuration

# settings.py
LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'celery': {
            'format': '[%(asctime)s: %(levelname)s/%(processName)s] [%(task_name)s(%(task_id)s)] %(message)s',
        },
    },
    'handlers': {
        'celery': {
            'level': 'INFO',
            'class': 'logging.handlers.RotatingFileHandler',
            'filename': '/var/log/celery/tasks.log',
            'maxBytes': 1024 * 1024 * 100,  # 100MB
            'backupCount': 10,
            'formatter': 'celery',
        },
    },
    'loggers': {
        'celery': {
            'handlers': ['celery'],
            'level': 'INFO',
            'propagate': False,
        },
        'celery.task': {
            'handlers': ['celery'],
            'level': 'INFO',
            'propagate': False,
        },
    },
}

Structured Logging

import structlog
from celery import shared_task

logger = structlog.get_logger()

@shared_task(bind=True)
def process_order(self, order_id):
    log = logger.bind(
        task_id=self.request.id,
        task_name=self.name,
        order_id=order_id,
    )

    log.info('starting_order_processing')

    try:
        order = Order.objects.get(id=order_id)
        log = log.bind(user_id=order.user_id)

        result = process(order)
        log.info('order_processed', result=result)
        return result

    except Order.DoesNotExist:
        log.error('order_not_found')
        raise
    except Exception as e:
        log.exception('order_processing_failed', error=str(e))
        raise

Celery Signals for Logging

from celery import signals
import logging

logger = logging.getLogger('celery.task')

@signals.task_prerun.connect
def task_prerun_handler(task_id, task, args, kwargs, **kw):
    logger.info(f'Task starting: {task.name}[{task_id}]')

@signals.task_postrun.connect
def task_postrun_handler(task_id, task, args, kwargs, retval, state, **kw):
    logger.info(f'Task completed: {task.name}[{task_id}] state={state}')

@signals.task_failure.connect
def task_failure_handler(task_id, exception, args, kwargs, traceback, einfo, **kw):
    logger.error(
        f'Task failed: {task_id}',
        exc_info=True,
        extra={'args': args, 'kwargs': kwargs}
    )

@signals.task_retry.connect
def task_retry_handler(request, reason, einfo, **kw):
    logger.warning(f'Task retrying: {request.id} reason={reason}')

Prometheus Metrics

Setup with celery-exporter

pip install celery-exporter

celery-exporter --broker-url=redis://localhost:6379/0 --port=9808

Queue Monitoring

from celery import current_app
from prometheus_client import Gauge

QUEUE_LENGTH = Gauge('celery_queue_length', 'Queue length', ['queue'])

def update_queue_metrics():
    """Update queue length metrics."""
    with current_app.connection() as conn:
        for queue in ['default', 'high-priority', 'low-priority']:
            try:
                length = conn.default_channel.queue_declare(
                    queue=queue, passive=True
                ).message_count
                QUEUE_LENGTH.labels(queue=queue).set(length)
            except Exception:
                pass

# Call periodically via Beat
@shared_task
def collect_queue_metrics():
    update_queue_metrics()

Health Checks

Worker Health Check

from celery import current_app

def check_celery_health():
    """Check if Celery workers are healthy."""
    try:
        # Ping workers
        result = current_app.control.ping(timeout=5)
        if not result:
            return {'healthy': False, 'error': 'No workers responded'}

        # Check active workers
        active = current_app.control.inspect().active()
        if not active:
            return {'healthy': False, 'error': 'No active workers'}

        return {
            'healthy': True,
            'workers': len(active),
            'worker_names': list(active.keys()),
        }
    except Exception as e:
        return {'healthy': False, 'error': str(e)}

Django Health Check Integration

# healthchecks.py
from health_check.backends import BaseHealthCheckBackend
from health_check.exceptions import HealthCheckException
from celery import current_app

class CeleryHealthCheck(BaseHealthCheckBackend):
    critical_service = True

    def check_status(self):
        try:
            result = current_app.control.ping(timeout=3)
            if not result:
                raise HealthCheckException('No Celery workers available')
        except Exception as e:
            raise HealthCheckException(str(e))

    def identifier(self):
        return 'Celery Workers'

# settings.py
INSTALLED_APPS = [
    ...
    'health_check',
    'health_check.contrib.celery',
]

Beat Health Check

from django.core.cache import cache
from django.utils import timezone

@shared_task
def beat_heartbeat():
    """Record Beat scheduler heartbeat."""
    cache.set('celery_beat_heartbeat', timezone.now().isoformat(), timeout=120)

def check_beat_health():
    """Check if Beat is running."""
    heartbeat = cache.get('celery_beat_heartbeat')
    if not heartbeat:
        return {'healthy': False, 'error': 'No heartbeat recorded'}

    last_beat = timezone.datetime.fromisoformat(heartbeat)
    age = (timezone.now() - last_beat).seconds

    if age > 120:
        return {'healthy': False, 'error': f'Heartbeat is {age}s old'}

    return {'healthy': True, 'last_heartbeat': heartbeat}

Debugging Stuck Tasks

Inspecting Workers

from celery import current_app

def inspect_workers():
    """Get detailed worker information."""
    inspect = current_app.control.inspect()

    return {
        'active': inspect.active(),      # Currently executing tasks
        'reserved': inspect.reserved(),   # Prefetched tasks
        'scheduled': inspect.scheduled(), # ETA/countdown tasks
        'stats': inspect.stats(),         # Worker statistics
    }

def get_active_tasks():
    """List all currently running tasks."""
    inspect = current_app.control.inspect()
    active = inspect.active()

    tasks = []
    for worker, task_list in (active or {}).items():
        for task in task_list:
            tasks.append({
                'worker': worker,
                'task_id': task['id'],
                'task_name': task['name'],
                'args': task['args'],
                'started': task.get('time_start'),
            })
    return tasks

Finding Long-Running Tasks

import time

def find_long_running_tasks(threshold_seconds=300):
    """Find tasks running longer than threshold."""
    inspect = current_app.control.inspect()
    active = inspect.active() or {}

    long_running = []
    now = time.time()

    for worker, tasks in active.items():
        for task in tasks:
            start_time = task.get('time_start')
            if start_time:
                duration = now - start_time
                if duration > threshold_seconds:
                    long_running.append({
                        'worker': worker,
                        'task_id': task['id'],
                        'task_name': task['name'],
                        'duration_seconds': int(duration),
                    })

    return long_running

Queue Inspection

def inspect_queues():
    """Get queue information."""
    from kombu import Connection

    with Connection(current_app.conf.broker_url) as conn:
        queues = {}
        for queue_name in ['default', 'high-priority', 'low-priority']:
            try:
                queue = conn.SimpleQueue(queue_name)
                queues[queue_name] = {
                    'length': len(queue),
                }
                queue.close()
            except Exception as e:
                queues[queue_name] = {'error': str(e)}

    return queues

Alerting

Slack Alerts

import requests
from celery import signals

SLACK_WEBHOOK_URL = 'https://hooks.slack.com/services/XXX/YYY/ZZZ'

def send_slack_alert(message, level='warning'):
    color = {'info': '#36a64f', 'warning': '#ff9800', 'error': '#f44336'}
    requests.post(SLACK_WEBHOOK_URL, json={
        'attachments': [{
            'color': color.get(level, '#808080'),
            'text': message,
        }]
    })

@signals.task_failure.connect
def alert_on_failure(sender, task_id, exception, **kwargs):
    message = f"Task `{sender.name}` failed: {exception}"
    send_slack_alert(message, level='error')

@signals.task_retry.connect
def alert_on_retry(sender, request, reason, **kwargs):
    if request.retries >= 3:  # Only alert after multiple retries
        message = f"Task `{sender.name}` retrying ({request.retries}): {reason}"
        send_slack_alert(message, level='warning')

PagerDuty Integration

import pypd
from celery import signals

pypd.api_key = 'your-api-key'

@signals.task_failure.connect
def pagerduty_alert(sender, task_id, exception, **kwargs):
    # Only page for critical tasks
    critical_tasks = ['process_payment', 'send_critical_notification']

    if sender.name in critical_tasks:
        pypd.Event.create(data={
            'service_key': 'your-service-key',
            'event_type': 'trigger',
            'description': f'Critical task failed: {sender.name}',
            'details': {
                'task_id': task_id,
                'exception': str(exception),
            }
        })

Dashboard Queries

Key Metrics to Monitor

Periodic Tasks with Celery Beat

Celery Beat Overview

Celery Beat is a scheduler that kicks off tasks at regular intervals. Tasks are executed by available workers.

Architecture

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│  Celery Beat    │────▶│  Message Broker │────▶│  Celery Worker  │
│  (scheduler)    │     │  (Redis/RMQ)    │     │  (executor)     │
└─────────────────┘     └─────────────────┘     └─────────────────┘

Important: Only run ONE Beat scheduler instance to avoid duplicate task execution.

Basic Configuration

Static Schedule in Settings

# settings.py
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    # Execute every 30 seconds
    'check-heartbeat': {
        'task': 'myapp.tasks.check_heartbeat',
        'schedule': 30.0,
    },

    # Execute every minute
    'process-queue': {
        'task': 'myapp.tasks.process_queue',
        'schedule': 60.0,
        'args': (),
    },

    # Execute at midnight every day
    'daily-cleanup': {
        'task': 'myapp.tasks.daily_cleanup',
        'schedule': crontab(hour=0, minute=0),
    },

    # Execute Monday at 7:30am
    'weekly-report': {
        'task': 'myapp.tasks.send_weekly_report',
        'schedule': crontab(hour=7, minute=30, day_of_week=1),
    },

    # Every hour on the hour
    'hourly-sync': {
        'task': 'myapp.tasks.sync_data',
        'schedule': crontab(minute=0),
    },
}

Schedule Types

from celery.schedules import crontab, solar
from datetime import timedelta

CELERY_BEAT_SCHEDULE = {
    # Timedelta - simple intervals
    'every-10-seconds': {
        'task': 'myapp.tasks.quick_check',
        'schedule': timedelta(seconds=10),
    },

    'every-5-minutes': {
        'task': 'myapp.tasks.periodic_check',
        'schedule': timedelta(minutes=5),
    },

    # Crontab - cron-like schedule
    'weekday-mornings': {
        'task': 'myapp.tasks.morning_task',
        'schedule': crontab(hour=8, minute=0, day_of_week='mon-fri'),
    },

    # First of every month
    'monthly-report': {
        'task': 'myapp.tasks.monthly_report',
        'schedule': crontab(day_of_month=1, hour=0, minute=0),
    },

    # Every quarter (1st of Jan, Apr, Jul, Oct)
    'quarterly-report': {
        'task': 'myapp.tasks.quarterly_report',
        'schedule': crontab(day_of_month=1, month_of_year='1,4,7,10', hour=0, minute=0),
    },

    # Solar schedule - based on sunrise/sunset
    'at-sunrise': {
        'task': 'myapp.tasks.sunrise_task',
        'schedule': solar('sunrise', -37.81753, 144.96715),  # Melbourne
    },
}

Crontab Reference

from celery.schedules import crontab

# Crontab arguments:
# minute, hour, day_of_week, day_of_month, month_of_year

# Every minute
crontab()

# Every hour at minute 0
crontab(minute=0)

# Every day at midnight
crontab(minute=0, hour=0)

# Every Monday
crontab(minute=0, hour=0, day_of_week=1)

# Every 15 minutes
crontab(minute='*/15')

# Every hour between 9am-5pm on weekdays
crontab(minute=0, hour='9-17', day_of_week='mon-fri')

# At 12:30 on the 15th of every month
crontab(minute=30, hour=12, day_of_month=15)

# Multiple specific values
crontab(minute=0, hour='0,12')  # Midnight and noon
crontab(day_of_week='mon,wed,fri')  # MWF

Django Database Scheduler

Setup

pip install django-celery-beat

# settings.py
INSTALLED_APPS = [
    ...
    'django_celery_beat',
]

# Use database scheduler
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

python manage.py migrate django_celery_beat

Managing Schedules via Admin

# admin.py
from django_celery_beat.models import (
    PeriodicTask,
    IntervalSchedule,
    CrontabSchedule,
)

# These are auto-registered, but you can customize:
from django.contrib import admin
from django_celery_beat.admin import PeriodicTaskAdmin
from django_celery_beat.models import PeriodicTask

class CustomPeriodicTaskAdmin(PeriodicTaskAdmin):
    list_display = ['name', 'task', 'enabled', 'last_run_at']
    list_filter = ['enabled', 'task']

admin.site.unregister(PeriodicTask)
admin.site.register(PeriodicTask, CustomPeriodicTaskAdmin)

Creating Schedules Programmatically

from django_celery_beat.models import (
    PeriodicTask,
    IntervalSchedule,
    CrontabSchedule,
)
import json

# Create interval schedule (every 10 minutes)
schedule, created = IntervalSchedule.objects.get_or_create(
    every=10,
    period=IntervalSchedule.MINUTES,
)

# Create periodic task with interval
PeriodicTask.objects.create(
    interval=schedule,
    name='Sync Data Every 10 Minutes',
    task='myapp.tasks.sync_data',
    args=json.dumps([]),
    kwargs=json.dumps({}),
)

# Create crontab schedule (daily at 6am)
cron_schedule, _ = CrontabSchedule.objects.get_or_create(
    minute='0',
    hour='6',
    day_of_week='*',
    day_of_month='*',
    month_of_year='*',
)

# Create periodic task with crontab
PeriodicTask.objects.create(
    crontab=cron_schedule,
    name='Daily Report',
    task='myapp.tasks.generate_daily_report',
    args=json.dumps(['pdf']),
    kwargs=json.dumps({'include_charts': True}),
    enabled=True,
)

Dynamic Schedule Management

from django_celery_beat.models import PeriodicTask, IntervalSchedule
import json

def create_user_notification_schedule(user_id, interval_minutes):
    """Create a periodic task for user-specific notifications."""
    schedule, _ = IntervalSchedule.objects.get_or_create(
        every=interval_minutes,
        period=IntervalSchedule.MINUTES,
    )

    task_name = f'notify-user-{user_id}'

    PeriodicTask.objects.update_or_create(
        name=task_name,
        defaults={
            'interval': schedule,
            'task': 'notifications.tasks.send_user_digest',
            'kwargs': json.dumps({'user_id': user_id}),
            'enabled': True,
        }
    )

def disable_user_notifications(user_id):
    """Disable notifications for a user."""
    task_name = f'notify-user-{user_id}'
    PeriodicTask.objects.filter(name=task_name).update(enabled=False)

def delete_user_notifications(user_id):
    """Remove notification schedule for a user."""
    task_name = f'notify-user-{user_id}'
    PeriodicTask.objects.filter(name=task_name).delete()

Timezone Handling

Configuration

# settings.py
CELERY_TIMEZONE = 'America/New_York'
CELERY_ENABLE_UTC = True  # Store in UTC, display in timezone

# Or use Django's timezone
CELERY_TIMEZONE = TIME_ZONE

Timezone-Aware Crontabs

from celery.schedules import crontab
import pytz

CELERY_BEAT_SCHEDULE = {
    # This runs at 9am in the configured CELERY_TIMEZONE
    'morning-email': {
        'task': 'myapp.tasks.send_morning_email',
        'schedule': crontab(hour=9, minute=0),
    },
}

# For database scheduler, set timezone on CrontabSchedule
from django_celery_beat.models import CrontabSchedule

schedule = CrontabSchedule.objects.create(
    minute='0',
    hour='9',
    timezone=pytz.timezone('America/New_York'),
)

Task Arguments and Options

Passing Arguments

CELERY_BEAT_SCHEDULE = {
    'daily-report': {
        'task': 'myapp.tasks.generate_report',
        'schedule': crontab(hour=6, minute=0),
        'args': ('daily', 'pdf'),
        'kwargs': {'include_charts': True, 'recipients': ['admin@example.com']},
    },
}

Task Options

CELERY_BEAT_SCHEDULE = {
    'priority-task': {
        'task': 'myapp.tasks.important_task',
        'schedule': crontab(minute='*/5'),
        'options': {
            'queue': 'high-priority',
            'priority': 9,
            'expires': 300,  # Expire if not started in 5 min
        },
    },
}

Preventing Duplicate Executions

Using Locks

from django.core.cache import cache
from celery import shared_task
from contextlib import contextmanager

@contextmanager
def beat_lock(lock_id, timeout=60*60):
    """Prevent duplicate periodic task execution."""
    lock_acquired = cache.add(lock_id, 'locked', timeout)
    try:
        yield lock_acquired
    finally:
        if lock_acquired:
            cache.delete(lock_id)

@shared_task
def hourly_sync():
    with beat_lock('hourly-sync-lock') as acquired:
        if not acquired:
            return  # Another instance is running

        # Safe to execute
        perform_sync()

Using celery-once

pip install celery-once

from celery_once import QueueOnce

@shared_task(base=QueueOnce, once={'graceful': True})
def slow_task():
    # Only one instance runs at a time
    perform_slow_operation()

Running Celery Beat

Development

# Run beat with worker (for development only)
celery -A myproject worker --beat --loglevel=info

# Run beat separately
celery -A myproject beat --loglevel=info

Production

# Run beat as separate process
celery -A myproject beat --loglevel=info --pidfile=/var/run/celery/beat.pid

# With database scheduler
celery -A myproject beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler

Best Practices

Schedule Design

# GOOD: Offset schedules to avoid thundering herd
CELERY_BEAT_SCHEDULE = {
    'task-a': {'task': 'app.task_a', 'schedule': crontab(minute=0)},  # :00
    'task-b': {'task': 'app.task_b', 'schedule': crontab(minute=5)},  # :05
    'task-c': {'task': 'app.task_c', 'schedule': crontab(minute=10)}, # :10
}

# BAD: All tasks at the same time
CELERY_BEAT_SCHEDULE = {
    'task-a': {'task': 'app.task_a', 'schedule': crontab(minute=0)},
    'task-b': {'task': 'app.task_b', 'schedule': crontab(minute=0)},
    'task-c': {'task': 'app.task_c', 'schedule': crontab(minute=0)},
}

Handling Long-Running Tasks

@shared_task(
    soft_time_limit=3600,  # 1 hour soft limit
    time_limit=3660,       # Hard limit slightly longer
)
def daily_report():
    """Generate daily report - may take up to 1 hour."""
    # Set schedule to run with enough gap
    pass

CELERY_BEAT_SCHEDULE = {
    'daily-report': {
        'task': 'myapp.tasks.daily_report',
        'schedule': crontab(hour=2, minute=0),  # 2am, plenty of time before business hours
        'options': {'expires': 3600 * 3},  # Expire if not started in 3 hours
    },
}

Monitoring Beat Health

@shared_task
def beat_health_check():
    """Record that beat is running."""
    cache.set('celery-beat-heartbeat', timezone.now(), timeout=120)

CELERY_BEAT_SCHEDULE = {
    'beat-health-check': {
        'task': 'myapp.tasks.beat_health_check',
        'schedule': 60.0,  # Every minute
    },
}

# Check beat health
def is_beat_healthy():
    last_heartbeat = cache.get('celery-beat-heartbeat')
    if not last_heartbeat:
        return False
    return (timezone.now() - last_heartbeat).seconds < 120

Cleanup Old Task Results

# Clean up task results periodically
@shared_task
def cleanup_task_results():
    """Remove old task results from database."""
    from django_celery_results.models import TaskResult

    cutoff = timezone.now() - timedelta(days=7)
    TaskResult.objects.filter(date_created__lt=cutoff).delete()

CELERY_BEAT_SCHEDULE = {
    'cleanup-results': {
        'task': 'myapp.tasks.cleanup_task_results',
        'schedule': crontab(hour=3, minute=0),  # Daily at 3am
    },
}

Common Patterns

Conditional Execution

@shared_task
def conditional_task():
    """Only execute if conditions are met."""
    if not should_run_today():
        return

    perform_task()

def should_run_today():
    # Skip weekends
    if timezone.now().weekday() >= 5:
        return False
    # Skip holidays
    if Holiday.objects.filter(date=timezone.now().date()).exists():
        return False
    return True

Chained Periodic Tasks

from celery import chain

@shared_task
def start_daily_pipeline():
    """Kick off daily processing pipeline."""
    workflow = chain(
        fetch_data.s(),
        process_data.s(),
        generate_report.s(),
        send_notifications.s(),
    )
    workflow.delay()

CELERY_BEAT_SCHEDULE = {
    'daily-pipeline': {
        'task': 'myapp.tasks.start_daily_pipeline',
        'schedule': crontab(hour=6, minute=0),
    },
}

Production Deployment

Worker Deployment Patterns

Basic Worker Command

celery -A myproject worker --loglevel=info

Worker Options

celery -A myproject worker \
    --loglevel=info \
    --concurrency=4 \
    --hostname=worker1@%h \
    --queues=default,high-priority \
    --prefetch-multiplier=1 \
    --max-tasks-per-child=1000 \
    --time-limit=300 \
    --soft-time-limit=240

Process Supervision

Systemd Configuration

# /etc/systemd/system/celery.service
[Unit]
Description=Celery Service
After=network.target

[Service]
Type=forking
User=celery
Group=celery
EnvironmentFile=/etc/conf.d/celery
WorkingDirectory=/opt/myproject
ExecStart=/bin/sh -c '${CELERY_BIN} -A ${CELERY_APP} multi start ${CELERYD_NODES} \
    --pidfile=${CELERYD_PID_FILE} \
    --logfile=${CELERYD_LOG_FILE} \
    --loglevel=${CELERYD_LOG_LEVEL} \
    ${CELERYD_OPTS}'
ExecStop=/bin/sh -c '${CELERY_BIN} multi stopwait ${CELERYD_NODES} \
    --pidfile=${CELERYD_PID_FILE}'
ExecReload=/bin/sh -c '${CELERY_BIN} -A ${CELERY_APP} multi restart ${CELERYD_NODES} \
    --pidfile=${CELERYD_PID_FILE} \
    --logfile=${CELERYD_LOG_FILE} \
    --loglevel=${CELERYD_LOG_LEVEL} \
    ${CELERYD_OPTS}'
Restart=always

[Install]
WantedBy=multi-user.target

# /etc/conf.d/celery
CELERY_BIN="/opt/myproject/venv/bin/celery"
CELERY_APP="myproject"
CELERYD_NODES="worker1 worker2"
CELERYD_OPTS="--time-limit=300 --concurrency=4"
CELERYD_PID_FILE="/var/run/celery/%n.pid"
CELERYD_LOG_FILE="/var/log/celery/%n%I.log"
CELERYD_LOG_LEVEL="INFO"

Systemd for Beat

# /etc/systemd/system/celerybeat.service
[Unit]
Description=Celery Beat Service
After=network.target

[Service]
Type=simple
User=celery
Group=celery
EnvironmentFile=/etc/conf.d/celery
WorkingDirectory=/opt/myproject
ExecStart=/bin/sh -c '${CELERY_BIN} -A ${CELERY_APP} beat \
    --pidfile=${CELERYBEAT_PID_FILE} \
    --logfile=${CELERYBEAT_LOG_FILE} \
    --loglevel=${CELERYD_LOG_LEVEL} \
    --scheduler django_celery_beat.schedulers:DatabaseScheduler'
Restart=always

[Install]
WantedBy=multi-user.target

Supervisor Configuration

# /etc/supervisor/conf.d/celery.conf
[program:celery-worker]
command=/opt/myproject/venv/bin/celery -A myproject worker --loglevel=INFO --concurrency=4
directory=/opt/myproject
user=celery
numprocs=1
stdout_logfile=/var/log/celery/worker.log
stderr_logfile=/var/log/celery/worker.log
autostart=true
autorestart=true
startsecs=10
stopwaitsecs=600
stopasgroup=true
priority=1000

[program:celery-beat]
command=/opt/myproject/venv/bin/celery -A myproject beat --loglevel=INFO --scheduler django_celery_beat.schedulers:DatabaseScheduler
directory=/opt/myproject
user=celery
numprocs=1
stdout_logfile=/var/log/celery/beat.log
stderr_logfile=/var/log/celery/beat.log
autostart=true
autorestart=true
startsecs=10
priority=999

Containerized Deployment

Dockerfile

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

# Create non-root user
RUN useradd -m celery
USER celery

# Default command (override in compose)
CMD ["celery", "-A", "myproject", "worker", "--loglevel=info"]

Docker Compose

# docker-compose.yml
version: '3.8'

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data

  web:
    build: .
    command: gunicorn myproject.wsgi:application --bind 0.0.0.0:8000
    volumes:
      - .:/app
    ports:
      - "8000:8000"
    environment:
      - CELERY_BROKER_URL=redis://redis:6379/0
      - CELERY_RESULT_BACKEND=redis://redis:6379/1
    depends_on:
      - redis

  celery-worker:
    build: .
    command: celery -A myproject worker --loglevel=info --concurrency=4
    volumes:
      - .:/app
    environment:
      - CELERY_BROKER_URL=redis://redis:6379/0
      - CELERY_RESULT_BACKEND=redis://redis:6379/1
    depends_on:
      - redis
    deploy:
      replicas: 2
      resources:
        limits:
          memory: 512M

  celery-beat:
    build: .
    command: celery -A myproject beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler
    volumes:
      - .:/app
    environment:
      - CELERY_BROKER_URL=redis://redis:6379/0
    depends_on:
      - redis
    deploy:
      replicas: 1  # Only one beat instance!

  flower:
    build: .
    command: celery -A myproject flower --port=5555
    ports:
      - "5555:5555"
    environment:
      - CELERY_BROKER_URL=redis://redis:6379/0
    depends_on:
      - redis

volumes:
  redis-data:

Kubernetes Deployment

# celery-worker-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: celery-worker
spec:
  replicas: 3
  selector:
    matchLabels:
      app: celery-worker
  template:
    metadata:
      labels:
        app: celery-worker
    spec:
      containers:
      - name: celery-worker
        image: myproject:latest
        command: ["celery", "-A", "myproject", "worker", "--loglevel=info"]
        env:
        - name: CELERY_BROKER_URL
          valueFrom:
            secretKeyRef:
              name: celery-secrets
              key: broker-url
        resources:
          limits:
            memory: "512Mi"
            cpu: "500m"
          requests:
            memory: "256Mi"
            cpu: "250m"
        livenessProbe:
          exec:
            command:
            - celery
            - -A
            - myproject
            - inspect
            - ping
          initialDelaySeconds: 30
          periodSeconds: 60
---
# celery-beat-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: celery-beat
spec:
  replicas: 1  # Must be exactly 1
  strategy:
    type: Recreate  # Prevent duplicate beats during deploy
  selector:
    matchLabels:
      app: celery-beat
  template:
    metadata:
      labels:
        app: celery-beat
    spec:
      containers:
      - name: celery-beat
        image: myproject:latest
        command: ["celery", "-A", "myproject", "beat", "--loglevel=info"]
        env:
        - name: CELERY_BROKER_URL
          valueFrom:
            secretKeyRef:
              name: celery-secrets
              key: broker-url

Scaling Strategies

Horizontal Scaling

# Scale workers based on queue length
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: celery-worker-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: celery-worker
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: External
    external:
      metric:
        name: celery_queue_length
      target:
        type: AverageValue
        averageValue: 100  # Scale up when > 100 tasks per worker

Queue-Based Scaling

# Dedicated workers for different queues
# High-priority: More workers, less concurrency (faster response)
celery -A myproject worker -Q high-priority -c 2 -n high@%h

# Default: Balanced
celery -A myproject worker -Q default -c 4 -n default@%h

# Low-priority: Fewer workers, more concurrency (batch processing)
celery -A myproject worker -Q low-priority -c 8 -n low@%h

Deployment Warnings

Task Signature Changes

Critical: Drain all queues before deploying changes to task signatures (adding, removing, or renaming parameters).

# BEFORE: Task with two parameters
@shared_task
def process_order(order_id, notify=True):
    pass

# AFTER: Task with changed signature
@shared_task
def process_order(order_id, notify=True, priority='normal'):  # New param
    pass

Problem: Old messages in the queue have the old signature. When new workers pick them up, they may fail or behave unexpectedly.

Solution:

# 1. Stop producers (or pause task creation)
# 2. Wait for queues to drain
celery -A myproject inspect reserved
celery -A myproject inspect active

# 3. Verify queues are empty
celery -A myproject inspect stats  # Check queue lengths

# 4. Deploy new code
# 5. Restart workers
# 6. Resume producers

Alternative: Backwards-Compatible Changes

# Make new parameters optional with defaults
@shared_task
def process_order(order_id, notify=True, priority=None):
    if priority is None:
        priority = 'normal'  # Handle old messages
    pass

Avoid Long ETAs and Countdowns

Tasks with long eta or countdown values can cause issues:

# BAD: Task scheduled far in the future
my_task.apply_async(eta=datetime.now() + timedelta(days=30))
my_task.apply_async(countdown=60*60*24*30)  # 30 days

# Problems:
# - Task signature might change before execution
# - Broker restarts may lose the task
# - Visibility timeout issues with Redis/SQS

Better alternatives:

• Use Celery Beat for scheduled tasks
• Store future tasks in database with a periodic recovery task
• Use a dedicated scheduling service

Soft Shutdown (Celery 5.5+)

Celery 5.5+ introduced soft shutdown for cleaner worker termination.

# settings.py
CELERY_WORKER_ENABLE_SOFT_SHUTDOWN_ON_IDLE = True
CELERY_WORKER_SOFT_SHUTDOWN_TIMEOUT = 60  # Wait up to 60s for tasks to complete

# Gracefully shutdown with soft timeout
celery -A myproject control shutdown --soft-timeout=60

Benefits:

• Workers finish current tasks before shutting down
• No task loss during deployments
• Cleaner rolling updates in Kubernetes

Graceful Shutdown

Signal Handling

# celery.py
from celery.signals import worker_shutting_down

@worker_shutting_down.connect
def worker_shutting_down_handler(sig, how, exitcode, **kwargs):
    logger.info(f'Worker shutting down: signal={sig}, how={how}')
    # Perform cleanup
    cleanup_connections()

Kubernetes Graceful Shutdown

spec:
  containers:
  - name: celery-worker
    lifecycle:
      preStop:
        exec:
          command: ["/bin/sh", "-c", "celery -A myproject control shutdown"]
    terminationGracePeriodSeconds: 300  # 5 minutes for tasks to complete

Docker Compose Graceful Shutdown

services:
  celery-worker:
    stop_grace_period: 5m  # Wait up to 5 minutes
    stop_signal: SIGTERM   # Send SIGTERM for graceful shutdown

Health Checks

Liveness Probe

# health.py
from celery import current_app

def celery_liveness():
    """Check if worker can receive tasks."""
    try:
        result = current_app.control.ping(timeout=5)
        return bool(result)
    except Exception:
        return False

# Kubernetes liveness probe
livenessProbe:
  exec:
    command:
    - python
    - -c
    - "from myproject.health import celery_liveness; exit(0 if celery_liveness() else 1)"
  initialDelaySeconds: 30
  periodSeconds: 30
  timeoutSeconds: 10

Readiness Probe

def celery_readiness():
    """Check if worker is ready to accept tasks."""
    try:
        inspect = current_app.control.inspect()
        stats = inspect.stats()
        return bool(stats)
    except Exception:
        return False

Production Settings Checklist

# settings/production.py

# Broker
CELERY_BROKER_URL = os.environ['CELERY_BROKER_URL']
CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP = True
CELERY_BROKER_CONNECTION_MAX_RETRIES = 10

# Result backend
CELERY_RESULT_BACKEND = os.environ.get('CELERY_RESULT_BACKEND', 'django-db')
CELERY_RESULT_EXPIRES = 60 * 60 * 24  # 24 hours

# Serialization (JSON only for security)
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']

# Task execution
CELERY_TASK_ACKS_LATE = True
CELERY_TASK_REJECT_ON_WORKER_LOST = True
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60  # 30 min
CELERY_TASK_SOFT_TIME_LIMIT = 25 * 60

# Worker
CELERY_WORKER_PREFETCH_MULTIPLIER = 1
CELERY_WORKER_MAX_TASKS_PER_CHILD = 1000
CELERY_WORKER_DISABLE_RATE_LIMITS = False

# Beat (single instance only)
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

# Monitoring
CELERY_SEND_EVENTS = True
CELERY_SEND_TASK_SENT_EVENT = True

Security Considerations

Network Security

# docker-compose - internal network
services:
  redis:
    networks:
      - internal
    # No ports exposed to host

  celery-worker:
    networks:
      - internal
    # Access redis via internal network

networks:
  internal:
    driver: bridge

Secrets Management

# Never hardcode credentials
# Use environment variables or secrets manager
import os

CELERY_BROKER_URL = os.environ['CELERY_BROKER_URL']

# Or use AWS Secrets Manager, HashiCorp Vault, etc.
from myproject.secrets import get_secret
CELERY_BROKER_URL = get_secret('celery/broker-url')

Task Argument Validation

@shared_task(bind=True)
def secure_task(self, user_id, action):
    # Validate inputs
    if not isinstance(user_id, int):
        raise ValueError("Invalid user_id type")
    if action not in ['activate', 'deactivate']:
        raise ValueError("Invalid action")

    # Process safely
    process_action(user_id, action)

Deployment Checklist

• [ ] Use process supervisor (systemd, supervisor, K8s)
• [ ] Only one Beat instance running
• [ ] Graceful shutdown configured
• [ ] Health checks in place
• [ ] Logging to persistent storage
• [ ] Monitoring and alerting configured
• [ ] Secrets properly managed
• [ ] Resource limits set
• [ ] Backup strategy for broker/backend
• [ ] Auto-scaling configured (if needed)
• [ ] Security groups/firewall rules configured
• [ ] SSL/TLS for broker connections (production)

Task Design Patterns

Task Fundamentals

Basic Task Definition

from celery import shared_task

# Simple task - use for reusable Django apps
@shared_task
def add(x, y):
    return x + y

# Bound task - access to self for retries, logging
@shared_task(bind=True)
def process_data(self, data_id):
    self.update_state(state='PROCESSING')
    # ...

Task Naming

Always use explicit names for production tasks:

# Explicit name - recommended for production
@shared_task(name='orders.tasks.process_order')
def process_order(order_id):
    pass

# Auto-generated name (module.function_name)
# Can break if you refactor or rename
@shared_task
def process_order(order_id):
    pass

Task Signatures

# Different ways to call tasks
task.delay(arg1, arg2)              # Shortcut for apply_async
task.apply_async(args=[arg1, arg2]) # Full control
task.apply_async(
    args=[arg1],
    kwargs={'key': 'value'},
    countdown=60,                    # Delay execution by 60 seconds
    eta=datetime(2024, 1, 1, 12, 0), # Execute at specific time
    expires=3600,                    # Expire if not started in 1 hour
    queue='high-priority',           # Route to specific queue
    priority=9,                      # Task priority (0-9, higher = more priority)
)

# Signature objects for building workflows
from celery import signature
sig = signature('tasks.add', args=(2, 2))
sig = add.s(2, 2)  # Shortcut
sig = add.si(2, 2) # Immutable signature (ignores return from previous task)

Argument Best Practices

Pass IDs, Not Objects

# BAD - Django objects aren't JSON serializable
@shared_task
def process_order(order):  # Don't pass model instances!
    order.process()

# GOOD - Pass primary keys
@shared_task
def process_order(order_id):
    from orders.models import Order
    order = Order.objects.get(id=order_id)
    order.process()

Handle Missing Objects

@shared_task(bind=True)
def send_notification(self, user_id):
    from users.models import User

    try:
        user = User.objects.get(id=user_id)
    except User.DoesNotExist:
        # Object was deleted between queuing and execution
        # Log and skip - don't retry
        logger.warning(f"User {user_id} not found, skipping notification")
        return None

    user.send_notification()

Serializable Arguments Only

# GOOD - primitive types
@shared_task
def process(user_id, action, options=None):
    pass

process.delay(123, 'activate', {'notify': True})

# BAD - non-serializable types
@shared_task
def process(user, callback_func):  # Can't serialize!
    pass

Idempotency

Why Idempotency Matters

Tasks may execute multiple times due to:

• Network issues causing duplicate messages
• Worker crashes mid-execution followed by retry
• Manual task replay for debugging

Idempotent Task Patterns

# BAD - Not idempotent, will increment multiple times on retry
@shared_task
def increment_counter(counter_id):
    counter = Counter.objects.get(id=counter_id)
    counter.value += 1
    counter.save()

# GOOD - Idempotent using state check
@shared_task
def increment_counter(counter_id, expected_value):
    Counter.objects.filter(
        id=counter_id,
        value=expected_value  # Only update if value hasn't changed
    ).update(value=F('value') + 1)

# GOOD - Idempotent using unique constraint
@shared_task
def create_order_item(order_id, product_id, quantity, idempotency_key):
    OrderItem.objects.get_or_create(
        idempotency_key=idempotency_key,
        defaults={
            'order_id': order_id,
            'product_id': product_id,
            'quantity': quantity,
        }
    )

Deduplication with Locks

from django.core.cache import cache
from contextlib import contextmanager

@contextmanager
def task_lock(lock_id, timeout=60*10):
    lock_acquired = cache.add(lock_id, 'locked', timeout)
    try:
        yield lock_acquired
    finally:
        if lock_acquired:
            cache.delete(lock_id)

@shared_task(bind=True)
def deduplicated_task(self, resource_id):
    lock_id = f'task-lock-{resource_id}'

    with task_lock(lock_id) as acquired:
        if not acquired:
            # Another task is processing this resource
            logger.info(f"Task for {resource_id} already running, skipping")
            return

        # Safe to process
        process_resource(resource_id)

Workflow Patterns

Chains - Sequential Execution

from celery import chain

# Each task passes its result to the next
workflow = chain(
    fetch_data.s(url),        # Returns data
    parse_data.s(),           # Receives data, returns parsed
    store_data.s(),           # Receives parsed, stores
    notify_complete.si()      # si() = immutable, ignores input
)
result = workflow.delay()

Groups - Parallel Execution

from celery import group

# Execute tasks in parallel
parallel = group(
    process_item.s(item_id) for item_id in item_ids
)
result = parallel.delay()

# Get all results (blocks until all complete)
results = result.get()

Chords - Parallel with Callback

from celery import chord

# Run tasks in parallel, then call callback with all results
workflow = chord(
    [process_item.s(item_id) for item_id in item_ids],
    aggregate_results.s()  # Called with list of all results
)
result = workflow.delay()

Complex Workflows

from celery import chain, group, chord

# Example: Process order with parallel item processing
workflow = chain(
    validate_order.s(order_id),
    chord(
        [process_item.s(item_id) for item_id in item_ids],
        finalize_order.s(order_id)
    ),
    send_confirmation.s()
)

Task Inheritance

Base Task Classes

from celery import Task

class DatabaseTask(Task):
    """Base task that ensures database connection."""

    def __call__(self, *args, **kwargs):
        from django.db import connection
        connection.ensure_connection()
        return super().__call__(*args, **kwargs)


class LoggedTask(Task):
    """Base task with automatic logging."""

    def on_success(self, retval, task_id, args, kwargs):
        logger.info(f'Task {self.name}[{task_id}] succeeded')

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        logger.error(f'Task {self.name}[{task_id}] failed: {exc}')

    def on_retry(self, exc, task_id, args, kwargs, einfo):
        logger.warning(f'Task {self.name}[{task_id}] retrying: {exc}')


@shared_task(base=LoggedTask, bind=True)
def my_logged_task(self, data):
    pass

Abstract Base Tasks

from celery import shared_task

class NotificationTaskMixin:
    """Mixin for notification tasks."""
    max_retries = 3
    default_retry_delay = 60

    def get_user(self, user_id):
        from users.models import User
        return User.objects.get(id=user_id)


class EmailNotificationTask(NotificationTaskMixin, Task):
    pass


@shared_task(base=EmailNotificationTask, bind=True)
def send_email_notification(self, user_id, template):
    user = self.get_user(user_id)
    # ...

Task Context and Metadata

Accessing Task Properties

@shared_task(bind=True)
def task_with_context(self, data):
    # Task metadata
    task_id = self.request.id
    task_name = self.name
    retries = self.request.retries

    # Execution context
    hostname = self.request.hostname
    delivery_info = self.request.delivery_info
    called_directly = self.request.called_directly

    # Parent task (if called from another task)
    parent_id = self.request.parent_id
    root_id = self.request.root_id  # Original task that started the chain

    # Headers (custom metadata)
    custom_header = self.request.headers.get('my_header')

Passing Context Through Chains

# Using headers for context that persists through chain
@shared_task(bind=True)
def step_one(self, data):
    correlation_id = self.request.headers.get('correlation_id')
    logger.info(f"[{correlation_id}] Processing step one")
    return data

# Apply with headers
chain(
    step_one.s(data),
    step_two.s(),
).apply_async(headers={'correlation_id': 'abc-123'})

Long-Running Tasks

Progress Tracking

@shared_task(bind=True)
def long_running_task(self, items):
    total = len(items)

    for i, item in enumerate(items):
        process_item(item)

        # Update progress
        self.update_state(
            state='PROGRESS',
            meta={
                'current': i + 1,
                'total': total,
                'percent': int((i + 1) / total * 100)
            }
        )

    return {'status': 'complete', 'processed': total}

# Check progress from Django view
def check_progress(request, task_id):
    result = long_running_task.AsyncResult(task_id)

    if result.state == 'PENDING':
        response = {'state': 'PENDING', 'progress': 0}
    elif result.state == 'PROGRESS':
        response = {
            'state': 'PROGRESS',
            'progress': result.info.get('percent', 0)
        }
    elif result.state == 'SUCCESS':
        response = {'state': 'SUCCESS', 'result': result.result}
    else:
        response = {'state': result.state, 'error': str(result.info)}

    return JsonResponse(response)

Chunking Large Datasets

from celery import chord, group

@shared_task
def process_chunk(item_ids):
    items = Item.objects.filter(id__in=item_ids)
    for item in items:
        item.process()
    return len(item_ids)

@shared_task
def aggregate_results(results):
    total = sum(results)
    logger.info(f"Processed {total} items total")
    return total

def process_all_items():
    all_ids = list(Item.objects.values_list('id', flat=True))
    chunk_size = 100

    # Split into chunks
    chunks = [
        all_ids[i:i + chunk_size]
        for i in range(0, len(all_ids), chunk_size)
    ]

    # Process chunks in parallel, aggregate at end
    workflow = chord(
        [process_chunk.s(chunk) for chunk in chunks],
        aggregate_results.s()
    )
    return workflow.delay()

Task Revocation

Revoking Tasks

from celery.result import AsyncResult

# Revoke a single task
result = my_task.delay(data)
result.revoke()

# Revoke with termination (kills running task)
result.revoke(terminate=True)

# Revoke by task ID
from myproject.celery import app
app.control.revoke(task_id, terminate=True)

# Revoke multiple tasks
app.control.revoke([task_id1, task_id2])

Handling Revocation in Tasks

from celery.exceptions import Terminated

@shared_task(bind=True)
def cancelable_task(self, items):
    for item in items:
        # Check if task was revoked
        if self.is_aborted():
            logger.info("Task was revoked, cleaning up")
            cleanup()
            return

        process_item(item)

Rate Limiting

Task-Level Rate Limits

# Limit to 10 tasks per minute
@shared_task(rate_limit='10/m')
def rate_limited_task(data):
    pass

# Limit to 100 tasks per hour
@shared_task(rate_limit='100/h')
def hourly_limited_task(data):
    pass

# Limit to 1 task per second
@shared_task(rate_limit='1/s')
def slow_task(data):
    pass

Dynamic Rate Limiting

from celery import shared_task
from time import sleep

@shared_task(bind=True)
def api_call_task(self, endpoint):
    try:
        response = call_api(endpoint)
        return response
    except RateLimitError as exc:
        # Back off and retry
        retry_after = exc.retry_after or 60
        raise self.retry(exc=exc, countdown=retry_after)