django-celery

intent

use this skill when you need background task processing in django. add celery + redis or rabbitmq for offloading slow work (email, pdfs, api calls) from the request cycle, scheduling periodic jobs with beat, retrying failed tasks with exponential backoff, chaining or grouping tasks with canvas, and testing tasks without a broker. production-grade patterns cover idempotency, soft timeouts, dead-letter queues, and monitoring.

inputs

python packages:

• celery[redis] or celery[amqp]
• django-celery-results
• django-celery-beat

environment variables:

• CELERY_BROKER_URL: redis://localhost:6379/0 or amqp://guest:guest@localhost//
• CELERY_RESULT_BACKEND: django-db, redis, or database-backed

external connections:

• redis or rabbitmq broker (network accessible from workers)
• sentry or similar (optional, for task failure monitoring)
• flower (optional, web ui for real-time monitoring)

django setup:

• config/celery.py entrypoint configured with app.autodiscover_tasks()
• INSTALLED_APPS includes django_celery_results and django_celery_beat
• settings keys: CELERY_BROKER_URL, CELERY_RESULT_BACKEND, CELERY_TASK_TIME_LIMIT, CELERY_TASK_SOFT_TIME_LIMIT, CELERY_TASK_ACKS_LATE, CELERY_BEAT_SCHEDULER

procedure

1. install and initialize celery

inputs: python project with django installed

steps:

• run pip install celery[redis] django-celery-results django-celery-beat
• create config/celery.py with app initialization:

  import os
  from celery import Celery
  
  os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.development')
  app = Celery('myproject')
  app.config_from_object('django.conf:settings', namespace='CELERY')
  app.autodiscover_tasks()
  

• update config/__init__.py to import the celery app:

  from .celery import app as celery_app
  __all__ = ('celery_app',)
  

outputs: celery app ready to discover and run tasks

2. configure django settings for celery

inputs: django settings module, redis or rabbitmq broker url

steps:

• set CELERY_BROKER_URL to redis or rabbitmq connection string
• set CELERY_RESULT_BACKEND to 'django-db' or a separate redis instance
• configure serialization: set CELERY_ACCEPT_CONTENT, CELERY_TASK_SERIALIZER, CELERY_RESULT_SERIALIZER to 'json' (safer than pickle)
• set timeouts: CELERY_TASK_TIME_LIMIT (hard kill, e.g. 30 min), CELERY_TASK_SOFT_TIME_LIMIT (graceful timeout, e.g. 25 min)
• set CELERY_TASK_ACKS_LATE = True to re-queue tasks if a worker crashes
• set CELERY_WORKER_PREFETCH_MULTIPLIER = 1 to prevent workers from hoarding long tasks
• set CELERY_RESULT_EXPIRES = 86400 (24 hours) to auto-clean task results
• add django_celery_results and django_celery_beat to INSTALLED_APPS
• set CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler' for db-backed schedules

outputs: django settings configured; celery will respect broker, serialization, and task behavior settings

3. start celery workers

inputs: django project with celery configured

steps for development:

• run celery -A config worker --loglevel=info to start a single worker consuming from 'default' queue

steps for production:

• run celery -A config worker --loglevel=warning --concurrency=4 -Q default,high_priority to start a worker with multiple concurrency and queue routing
• use supervisord, systemd, or container orchestration to restart workers on crash
• never run celery worker --beat in production; always run beat on a separate single-node service

outputs: worker process listening on broker, ready to consume and execute tasks

4. start celery beat scheduler

inputs: django project with django_celery_beat installed

steps:

• run celery -A config beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler to start beat on a single node
• (do not run beat on multiple nodes or use --beat flag in worker; will cause duplicate task executions)

outputs: beat process reading schedules from db, enqueuing periodic tasks at configured times

5. define a basic task

inputs: django app with tasks.py module

steps:

• create apps/notifications/tasks.py
• import shared_task: from celery import shared_task
• decorate a function with @shared_task(name='notifications.send_welcome_email')
• pass only primitives and pks, never orm instances
• handle missing objects gracefully without raising (idempotent pattern)
• log outcomes with logger.info or logger.warning

example:

from celery import shared_task
import logging

logger = logging.getLogger(__name__)

@shared_task(name='notifications.send_welcome_email')
def send_welcome_email(user_id: int) -> None:
    from apps.users.models import User
    from apps.notifications.services import EmailService
    
    try:
        user = User.objects.get(pk=user_id)
    except User.DoesNotExist:
        logger.warning('send_welcome_email: user %s not found', user_id)
        return
    
    EmailService.send_welcome(user)
    logger.info('Welcome email sent to user %s', user_id)

outputs: task function callable via task_name.delay() or .apply_async()

6. define a retryable task with backoff

inputs: task that may fail transiently (api calls, external service syncs)

steps:

• use @shared_task(bind=True, max_retries=5, default_retry_delay=60, autoretry_for=(ConnectionError, TimeoutError), retry_backoff=True, retry_backoff_max=600, retry_jitter=True)
• catch specific exceptions and call self.retry(exc=exc, countdown=...) to retry with exponential backoff
• for rate limits, extract retry_after header from response and pass as countdown

example:

@shared_task(
    bind=True,
    name='integrations.sync_to_crm',
    max_retries=5,
    default_retry_delay=60,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
)
def sync_contact_to_crm(self, contact_id: int) -> dict:
    from apps.crm.services import CRMClient
    
    try:
        result = CRMClient().sync(contact_id)
        return result
    except CRMClient.RateLimitError as exc:
        raise self.retry(exc=exc, countdown=int(exc.retry_after))

outputs: task retries up to max_retries with exponential backoff; jitter prevents thundering herd

7. implement idempotent task with status guard

inputs: task that modifies state and may be retried multiple times

steps:

• use select_for_update() to acquire a lock on the record
• check a status field to ensure the task has not already run (e.g., only update if status == PENDING)
• use conditional update: Order.objects.filter(pk=order_id, status=PENDING).update(status=SHIPPED)
• if update returns 0 rows, log and return without raising

example:

@shared_task(name='orders.mark_shipped')
def mark_order_shipped(order_id: int, tracking_number: str) -> None:
    from apps.orders.models import Order
    
    updated = Order.objects.filter(
        pk=order_id,
        status=Order.Status.PROCESSING,
    ).update(
        status=Order.Status.SHIPPED,
        tracking_number=tracking_number,
    )
    
    if not updated:
        logger.info('mark_order_shipped: order %s already shipped or not found', order_id)

outputs: task is safe to execute multiple times; no duplicate state changes

8. handle soft timeouts gracefully

inputs: long-running task (pdf generation, report building)

steps:

• set both soft_time_limit (e.g. 120 seconds) and time_limit (e.g. 150 seconds) on the task
• wrap task body in try/except catching SoftTimeLimitExceeded
• in except block, clean up partial files or resources before the hard kill
• re-raise the exception to allow celery to mark the task as failed

example:

from celery.exceptions import SoftTimeLimitExceeded

@shared_task(
    bind=True,
    name='reports.generate_pdf',
    soft_time_limit=120,
    time_limit=150,
)
def generate_pdf_report(self, report_id: int) -> str:
    from apps.reports.services import PDFGenerator
    
    try:
        path = PDFGenerator.build(report_id)
        return path
    except SoftTimeLimitExceeded:
        PDFGenerator.cleanup(report_id)
        raise

outputs: task cleans up partial state on timeout; no dangling resources

9. call a task from django code

inputs: task defined, celery worker running

steps:

• call task_name.delay(*args, **kwargs) for fire-and-forget async execution
• call task_name.apply_async(args=[...], countdown=3600) to schedule in the future
• call task_name.apply_async(args=[...], eta=timezone.now() + timedelta(days=1)) to schedule at a specific datetime
• call task_name.apply_async(args=[...], queue='high_priority') to route to a specific queue
• call task_name.apply(args=[...]) only in tests or debugging (runs synchronously, no broker)

example:

from celery import current_app
from datetime import timedelta
from django.utils import timezone

send_welcome_email.delay(user.pk)
send_reminder.apply_async(args=[user.pk], countdown=3600)
send_reminder.apply_async(args=[user.pk], eta=timezone.now() + timedelta(days=1))
sync_contact_to_crm.apply_async(args=[contact.pk], queue='high_priority')

outputs: task enqueued on broker; worker will execute asynchronously

10. define periodic tasks via code-based beat schedule

inputs: django settings, celery configured

steps:

• add CELERY_BEAT_SCHEDULE dict to django settings
• for each periodic task, define a dict entry with 'task' (task name), 'schedule' (crontab or int seconds), and optional 'args' or 'kwargs'
• use crontab(hour=2, minute=0) for 2am daily, crontab(day_of_week='monday', hour=8, minute=0) for mondays at 8am, or 60.0 for every 60 seconds
• beat will read this dict and enqueue tasks at the specified times

example:

from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'cleanup-expired-sessions': {
        'task': 'users.cleanup_expired_sessions',
        'schedule': crontab(hour=2, minute=0),
    },
    'sync-inventory': {
        'task': 'products.sync_inventory',
        'schedule': 60.0,
    },
    'weekly-digest': {
        'task': 'notifications.send_weekly_digest',
        'schedule': crontab(day_of_week='monday', hour=8, minute=0),
    },
}

outputs: beat enqueues tasks on schedule; no manual triggering needed

11. define periodic tasks via database (dynamic)

inputs: django_celery_beat installed, CELERY_BEAT_SCHEDULER = DatabaseScheduler

steps:

• create a CrontabSchedule via orm: CrontabSchedule.objects.get_or_create(hour='*/6', minute='0', timezone='UTC')
• create or update a PeriodicTask with the crontab, task name, and args (as json string)
• set 'enabled': True to activate; False to pause
• beat reads from db periodically (every 5 seconds by default) and enqueues tasks

example:

from django_celery_beat.models import PeriodicTask, CrontabSchedule
import json

schedule, _ = CrontabSchedule.objects.get_or_create(
    hour='*/6', minute='0', timezone='UTC',
)

PeriodicTask.objects.update_or_create(
    name='Sync inventory every 6 hours',
    defaults={
        'crontab': schedule,
        'task': 'products.sync_inventory',
        'args': json.dumps([]),
        'enabled': True,
    }
)

outputs: periodic task registered in db; beat will enqueue at scheduled intervals

12. chain tasks sequentially

inputs: multiple tasks that depend on each other's output

steps:

• import chain from celery
• build a chain: chain(task1.s(arg1), task2.s(), task3.s())
• each task receives the previous task's return value as its first positional argument
• call .delay() to enqueue the chain
• chain runs sequentially on a single worker; slower than parallel but preserves order and data flow

example:

from celery import chain

pipeline = chain(
    fetch_data.s(source_id),
    transform_data.s(),
    load_to_warehouse.s(),
)
pipeline.delay()

outputs: tasks execute sequentially; result of each task passed to the next

13. group tasks in parallel

inputs: multiple independent tasks

steps:

• import group from celery
• build a group: group(task1.s(arg1) for arg1 in arg_list)
• call .delay() to enqueue all tasks to the same queue
• workers execute tasks in parallel
• group does not wait or aggregate results

example:

from celery import group

parallel = group(
    send_welcome_email.s(user_id)
    for user_id in new_user_ids
)
parallel.delay()

outputs: all tasks enqueued; workers execute in parallel

14. chord tasks with callback

inputs: multiple tasks that must complete before a final aggregation task

steps:

• import chord from celery
• build a chord: chord(group(task1.s(arg), ...), callback_task.s())
• workers execute all tasks in the group in parallel
• when all complete, celery invokes the callback task with a list of results
• callback aggregates or processes the results

example:

from celery import chord, group

result = chord(
    group(process_chunk.s(chunk) for chunk in data_chunks),
    aggregate_results.s(),
)
result.delay()

outputs: parallel tasks run; callback invoked with all results once group completes

15. set up sentry error tracking for failed tasks

inputs: sentry project, celery task_failure signal

steps:

• install sentry: pip install sentry-sdk
• import sentry_sdk and task_failure signal in a task module (e.g., apps/core/tasks.py)
• connect a handler to @task_failure.connect that captures the exception with task context
• in the handler, create a sentry scope with task name, task_id, args, and kwargs
• call sentry_sdk.capture_exception(exception)

example:

from celery.signals import task_failure
import sentry_sdk

@task_failure.connect
def on_task_failure(sender, task_id, exception, args, kwargs, traceback, einfo, **kw):
    with sentry_sdk.new_scope() as scope:
        scope.set_context('celery', {
            'task': sender.name,
            'task_id': task_id,
            'args': args,
            'kwargs': kwargs,
        })
        sentry_sdk.capture_exception(exception)

outputs: all task failures logged to sentry with full context

16. route failed tasks to dead-letter queue

inputs: task that may fail after max retries; FailedTask or similar model for persistence

steps:

• define a task with max_retries
• on final exception (after retries exhausted), do not raise
• instead, persist the failure to a dead-letter table (FailedCharge, FailedPayment, etc.) with order_id, error message, and task_id
• return gracefully so the task is not retried further
• separately, query the dead-letter table for manual review or async retry

example:

@shared_task(bind=True, max_retries=3, name='payments.charge_card')
def charge_card(self, order_id: int) -> None:
    from apps.payments.models import Order, FailedCharge
    
    try:
        _do_charge(order_