Django Cheatsheet 09 — Celery

uv add celery redis

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

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")

app = Celery("myapp")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

# config/__init__.py
from .celery import app as celery_app
__all__ = ["celery_app"]

# settings.py
CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/0"
CELERY_TASK_TIME_LIMIT = 300
CELERY_TASK_SOFT_TIME_LIMIT = 270
CELERY_TASK_ACKS_LATE = True
CELERY_WORKER_PREFETCH_MULTIPLIER = 1

# blog/tasks.py
from celery import shared_task
import logging

log = logging.getLogger(__name__)

@shared_task
def send_email_async(to, subject, body):
    log.info("sending to %s", to)
    send_email(to, subject, body)

# Async (in queue)
send_email_async.delay("[email protected]", "subj", "body")

# More control
send_email_async.apply_async(
    args=["[email protected]", "subj", "body"],
    countdown=60,                   # delay 60s
    eta=datetime(2026, 1, 1),       # run at specific time
    queue="emails",
    priority=10,
)

celery -A config worker -l info
celery -A config worker -l info -Q emails,default
celery -A config worker -l info --concurrency=4

@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def flaky_task(self, x):
    try:
        do_something(x)
    except Exception as e:
        raise self.retry(exc=e, countdown=2 ** self.request.retries * 60)

@shared_task(
    autoretry_for=(ConnectionError, TimeoutError),
    max_retries=5,
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
)
def task():
    ...

@shared_task
def charge(payment_id, idempotency_key):
    p = Payment.objects.get(id=payment_id)
    if Charge.objects.filter(idempotency_key=idempotency_key).exists():
        return                       # already charged
    Charge.objects.create(payment_id=payment_id, idempotency_key=idempotency_key, ...)

uv add django-celery-beat

INSTALLED_APPS = [..., "django_celery_beat"]
CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers:DatabaseScheduler"

celery -A config beat -l info

from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    "cleanup-old-logs": {
        "task": "blog.tasks.cleanup_logs",
        "schedule": crontab(hour=2, minute=0),
    },
    "every-minute": {
        "task": "blog.tasks.poll",
        "schedule": 60.0,
    },
}

from celery import group, chain, chord

# Sequential
chain(task1.s(), task2.s(), task3.s())()

# Parallel
group(task.s(x) for x in items)()

# Map-reduce
chord((task.s(x) for x in items), reducer.s())()

result = task.delay(...)

result.get()                    # block until done (don't in views!)
result.get(timeout=10)
result.ready()
result.successful()
result.failed()
result.id

from celery.signals import task_failure, task_success

@task_failure.connect
def handle_failure(sender=None, task_id=None, exception=None, **kwargs):
    log.error("task failed", task_id=task_id, exc=exception)

uv add flower
celery -A config flower

@shared_task
def send_email_task(to, subject, body):
    send_mail(subject, body, "[email protected]", [to])

# In view:
send_email_task.delay(user.email, "Welcome", "Hi!")

@shared_task
def cleanup_expired():
    Session.objects.filter(expires_at__lt=timezone.now()).delete()

# Beat schedule daily at 3am

from django.db import transaction

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

@shared_task(bind=True)
def process(self, items):
    for i, item in enumerate(items):
        do(item)
        self.update_state(state="PROGRESS", meta={"done": i + 1, "total": len(items)})

result = process.delay(items)
# Poll: result.state, result.info