Django + Celery: Background Tasks and Task Queues

Some operations don’t belong in the request-response cycle. Sending emails, generating PDFs, processing uploads, syncing with external APIs — these should happen in the background. Celery is the standard solution for Django. It’s a distributed task queue that lets you offload work to separate worker processes, with Redis or RabbitMQ as the message broker.

Installation and Project Setup

Install Celery and the Redis backend:

pip install celery[redis] django-celery-results

Create the Celery app configuration alongside your Django settings:

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

# Set the default Django settings module
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")

app = Celery("myproject")

# Load config from Django settings, using the CELERY_ namespace
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}")

Make sure Django loads the Celery app on startup:

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

__all__ = ("celery_app",)

Add the Celery settings to your Django settings file:

# myproject/settings.py
CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "django-db"  # Store results in your database
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERY_TIMEZONE = "UTC"

INSTALLED_APPS += ["django_celery_results"]

Run the migration for the results backend:

python manage.py migrate django_celery_results

Writing Your First Task

Tasks live in tasks.py files inside your Django apps. Celery’s autodiscover_tasks() finds them automatically.

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


@shared_task
def send_welcome_email(user_id):
    """Send a welcome email in the background."""
    from django.contrib.auth.models import User

    user = User.objects.get(id=user_id)
    send_mail(
        subject="Welcome!",
        message=f"Hi {user.first_name}, thanks for signing up.",
        from_email="noreply@example.com",
        recipient_list=[user.email],
    )

Use @shared_task instead of @app.task so the task works regardless of which Celery app instance is active — important for reusable apps.

Always pass IDs, not objects. Celery serializes arguments to JSON. Pass user_id, not the User instance. The worker fetches a fresh copy from the database, avoiding stale data and serialization issues.

Calling Tasks

# In a view or service
from notifications.tasks import send_welcome_email

def register_user(request):
    user = User.objects.create_user(...)

    # Queue the task — returns immediately
    send_welcome_email.delay(user.id)

    return redirect("dashboard")

.delay() is shorthand for .apply_async(). For more control:

# Delay execution by 60 seconds
send_welcome_email.apply_async(args=[user.id], countdown=60)

# Execute at a specific time
from datetime import datetime, timezone
send_welcome_email.apply_async(args=[user.id], eta=datetime(2026, 7, 3, 9, 0, tzinfo=timezone.utc))

# Route to a specific queue
send_welcome_email.apply_async(args=[user.id], queue="emails")

Task Retries and Error Handling

Network failures, API rate limits, and transient errors are normal. Configure retries:

@shared_task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,  # seconds
)
def sync_to_crm(self, customer_id):
    """Sync customer data to an external CRM."""
    from .models import Customer
    import requests

    customer = Customer.objects.get(id=customer_id)

    try:
        response = requests.post(
            "https://api.crm.example.com/contacts",
            json={"email": customer.email, "name": customer.name},
            timeout=10,
        )
        response.raise_for_status()
    except requests.RequestException as exc:
        # Retry with exponential backoff
        raise self.retry(exc=exc, countdown=2 ** self.request.retries * 60)

The bind=True parameter passes the task instance as self, giving access to self.retry() and self.request. Exponential backoff prevents hammering a failing service.

For automatic retries on specific exceptions:

@shared_task(autoretry_for=(ConnectionError, TimeoutError), retry_backoff=True, max_retries=5)
def fetch_exchange_rates():
    """Autoretry with exponential backoff on connection errors."""
    import requests
    response = requests.get("https://api.exchange.example.com/rates", timeout=5)
    response.raise_for_status()
    # Process rates...

Task Design Patterns

Chaining Tasks

Run tasks in sequence where each result feeds the next:

from celery import chain

workflow = chain(
    download_report.s(report_id),
    parse_report.s(),           # receives result of download_report
    send_report_email.s(user_id),
)
workflow.apply_async()

Grouping Tasks

Run tasks in parallel and collect results:

from celery import group

job = group(
    process_image.s(image_id)
    for image_id in image_ids
)
result = job.apply_async()

Idempotent Tasks

Tasks can run more than once (broker redelivery, worker crash). Design them to be idempotent:

@shared_task
def charge_order(order_id):
    from .models import Order

    order = Order.objects.select_for_update().get(id=order_id)

    if order.payment_status == "charged":
        return  # Already processed — safe to skip

    process_payment(order)
    order.payment_status = "charged"
    order.save(update_fields=["payment_status"])

Periodic Tasks with Celery Beat

Celery Beat is a scheduler that sends tasks at regular intervals. Add it to your settings:

# myproject/settings.py
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    "cleanup-expired-sessions": {
        "task": "accounts.tasks.cleanup_expired_sessions",
        "schedule": crontab(hour=3, minute=0),  # Daily at 3 AM
    },
    "sync-exchange-rates": {
        "task": "payments.tasks.fetch_exchange_rates",
        "schedule": 3600.0,  # Every hour (in seconds)
    },
    "send-weekly-digest": {
        "task": "notifications.tasks.send_weekly_digest",
        "schedule": crontab(hour=9, minute=0, day_of_week=1),  # Monday at 9 AM
    },
}

Run the Beat scheduler alongside your worker:

# Terminal 1: Worker
celery -A myproject worker --loglevel=info

# Terminal 2: Beat scheduler
celery -A myproject beat --loglevel=info

# Or combine them (development only)
celery -A myproject worker --beat --loglevel=info

Running and Monitoring Workers

Start a worker:

celery -A myproject worker --loglevel=info --concurrency=4

For production, run workers with a process manager like systemd or supervisord:

# /etc/supervisor/conf.d/celery_worker.conf
[program:celery_worker]
command=/path/to/venv/bin/celery -A myproject worker --loglevel=info --concurrency=4
directory=/path/to/project
user=www-data
autostart=true
autorestart=true
stdout_logfile=/var/log/celery/worker.log
stderr_logfile=/var/log/celery/worker_error.log

Monitoring with Flower

Flower is a real-time web monitor for Celery:

pip install flower
celery -A myproject flower --port=5555

Open http://localhost:5555 to see active workers, task history, success/failure rates, and task details. In production, put Flower behind authentication.

Testing Tasks

Test tasks synchronously by calling the function directly, not through .delay():

from django.test import TestCase, override_settings
from notifications.tasks import send_welcome_email


class TaskTest(TestCase):
    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
    def test_welcome_email_task(self):
        """Run the task synchronously in tests."""
        result = send_welcome_email(self.user.id)
        # Assert email was sent
        from django.core import mail
        self.assertEqual(len(mail.outbox), 1)

CELERY_TASK_ALWAYS_EAGER=True runs tasks inline without a broker — perfect for unit tests.

Summary

Celery with Redis gives Django a production-ready task queue. Define tasks in tasks.py, call them with .delay(), handle failures with retries and exponential backoff, and schedule recurring work with Beat. Keep tasks idempotent, pass IDs instead of objects, and monitor everything with Flower.