Django Cheatsheet 17 — Background Tasks Patterns

Background tasks patterns cheatsheet.

When to use a queue

Move out of request cycle:

Email / SMS / push.
Image / video processing.
LLM calls.
Third-party API calls that may be slow.
Scheduled jobs.
Batch operations.

Don’t move:

Simple DB writes.
Quick computations.
Sync user-facing operations that must complete.

Choosing

	Celery	RQ	dramatiq	django-tasks (Django 5.1+)
Maturity	very high	high	high	new
Brokers	Redis, RabbitMQ, SQS	Redis	Redis, RabbitMQ	DB / Redis / custom
Schedule	Celery Beat	rq-scheduler	dramatiq.periodiq	django-celery-beat-like
Complexity	high	low	medium	very low

Default: Celery for big projects, RQ for small.

RQ (simpler alternative)

uv add rq django-rq

INSTALLED_APPS = [..., "django_rq"]

RQ_QUEUES = {
    "default": {
        "HOST": "localhost",
        "PORT": 6379,
        "DB": 0,
        "DEFAULT_TIMEOUT": 360,
    },
}

# tasks.py
import django_rq

@django_rq.job
def send_email(to, body):
    ...

# Usage:
send_email.delay("[email protected]", "hi")

uv run python manage.py rqworker default

dramatiq

uv add dramatiq[redis] django-dramatiq

import dramatiq

@dramatiq.actor
def send_email(to, body):
    ...

send_email.send("[email protected]", "hi")

Simpler than Celery, automatic retries.

django-tasks (Django 5.1+)

from django.tasks import default_task_backend, task

@task
def send_email(to, body):
    ...

send_email.enqueue("[email protected]", "hi")

Pluggable backends (Redis, DB, etc).

Patterns

Fire and forget

def view(request):
    send_email.delay(user.email, "Welcome")
    return redirect("/")

Confirmation pattern

def view(request):
    task = process.delay(...)
    return render(request, "processing.html", {"task_id": task.id})

def status(request, task_id):
    result = AsyncResult(task_id)
    return JsonResponse({"state": result.state, "info": result.info})

Frontend polls /status/<id>/.

Chain workflow

from celery import chain

chain(
    fetch.s(url),
    parse.s(),
    save.s(),
)()

Fan-out / fan-in

from celery import group, chord

# Process items in parallel, then aggregate
chord(
    (process.s(item) for item in items),
    aggregate.s(),
)()

Periodic

# Celery Beat
CELERY_BEAT_SCHEDULE = {
    "daily-cleanup": {
        "task": "blog.tasks.cleanup",
        "schedule": crontab(hour=2),
    },
}

Cron-like with django-cron / django-q2

Alternative to beat.

Retries

@shared_task(
    autoretry_for=(IOError, TimeoutError),
    max_retries=5,
    retry_backoff=True,           # 1s, 2s, 4s, 8s, ...
    retry_backoff_max=600,
    retry_jitter=True,
)
def fetch_data():
    ...

Always cap retries. Exponential backoff with jitter prevents thundering herd.

Dead-letter queue

# After max retries
@shared_task(bind=True, max_retries=3)
def process(self, id):
    try:
        do(id)
    except Exception as e:
        try:
            raise self.retry(exc=e)
        except MaxRetriesExceededError:
            DeadLetter.objects.create(task="process", payload=id, error=str(e))

Idempotency

Tasks may run twice (at-least-once delivery):

@shared_task
def charge(payment_id, idempotency_key):
    if Charge.objects.filter(idempotency_key=idempotency_key).exists():
        return
    ...

transaction.on_commit

def view(request):
    with transaction.atomic():
        order = Order.objects.create(...)
        transaction.on_commit(lambda: process_order.delay(order.id))

Without on_commit, the worker may pick up the task before the row is committed.

Result handling

# Sync polling (don't do in views!)
result = task.delay(...)
result.get(timeout=30)

# Webhook pattern
@shared_task
def process(payload):
    result = compute(payload)
    notify_webhook(payload["callback_url"], result)

Monitoring

Flower (Celery): web UI for tasks.
Sentry: captures task exceptions.
Grafana / Prometheus: queue depth, latency.

import sentry_sdk
sentry_sdk.init(integrations=[CeleryIntegration(), DjangoIntegration()])

Task discovery

# tasks.py — autodiscovery looks here
@shared_task
def my_task(...): ...

Or explicit:

app.autodiscover_tasks(["blog", "accounts"])

Common mistakes

Long-running tasks without time_limit → workers stuck.
Importing models inside a task — fine; but importing huge libs delays startup.
Tasks doing 1 transaction with 1 task call (overhead) → batch.
Same task accepting Model instance — not serializable. Pass IDs.
No idempotency + at-least-once → duplicate side effects.

When to use a queue#

Choosing#

RQ (simpler alternative)#

dramatiq#

django-tasks (Django 5.1+)#

Patterns#

Fire and forget#

Confirmation pattern#

Chain workflow#

Fan-out / fan-in#

Periodic#

Cron-like with django-cron / django-q2#

Retries#

Dead-letter queue#

Idempotency#

transaction.on_commit#

Result handling#

Monitoring#

Task discovery#

Common mistakes#

Read this next#