Background Tasks with Celery

Reviewed & published by Brayan K

Celery is a distributed task queue that runs slow or unreliable work in separate worker processes, so a Django view can enqueue a job and return immediately instead of making the user wait for it.

Part of the free Django course at LearnCodingFast — hands-on lessons with worked examples and the output they print, plus practice exercises and a quick quiz.

In this lesson you'll define a task with @shared_task, enqueue it with .delay(), understand the broker and result backend, run a worker that executes queued tasks, and add retries and periodic schedules with Beat.

What You'll Learn in This Lesson

• Defining a task with @shared_task
• Enqueueing work with .delay() instead of running it
• Brokers: Redis and RabbitMQ
• How a worker pulls and runs tasks
• The result backend for return values
• Retries and periodic tasks with Beat

1 Defining a Task and Calling .delay()

You mark a function as a Celery task with the @shared_task decorator. The crucial idea is that calling task.delay(args) does not run the function in your web process — it serializes the arguments into a message, puts that message on the broker, and returns instantly. A separate worker runs it later. This is how a view replies to the user in milliseconds while the slow work happens in the background.

# myapp/tasks.py
from celery import shared_task

@shared_task
def send_welcome_email(user_id):
    user = User.objects.get(pk=user_id)
    send_mail("Welcome!", "Thanks for joining.", "noreply@site.com", [user.email])
    return "sent"

# views.py — enqueue, then return immediately
def signup(request):
    user = create_user(request)
    send_welcome_email.delay(user.id)   # queued, NOT run here
    return redirect("home")             # user gets a fast response

# @shared_task turns a function into a task. Calling task.delay(args) does
# NOT run it now -- it ENQUEUES a message for a worker to run later. This
# simulates the broker queue: delay() pushes, the worker pops and runs.

class TaskQueue:
    def __init__(self):
        self.messages = []          # this is the "broker" (Redis/RabbitMQ)
    def enqueue(self, func, args):
        self.messages.append((func, args))
        return f"task-id-{len(self.messages)}"   # like an AsyncResult id

broker = TaskQueue()

def shared_task(func):
    # attach a .delay() that enqueues instead of running
    def delay(*args):
        return broker.enqueue(func, args)
    func.delay = delay
    return func

@shared_task
def send_email(to):
    return f"email sent to {to}"

# Calling .delay() returns immediately with a task id -- nothing ran yet
task_id = send_email.delay("user@example.com")
print("delay() returned task id:", task_id)
print("queued messages:", len(broker.messages))
print("(the email has NOT been sent yet -- a worker will run it)")

# EXPECTED:
# delay() returned task id: task-id-1
# queued messages: 1
# (the email has NOT been sent yet -- a worker will run it)

# ✅ Expected output:
# delay() returned task id: task-id-1
# queued messages: 1
# (the email has NOT been sent yet -- a worker will run it)

2 Brokers, Workers, and the Result Backend

Three pieces make Celery work. The broker (Redis or RabbitMQ) holds the queued messages. A worker is a separate process you start with celery worker; it pulls messages off the broker and runs the tasks. The optional result backend stores each task's return value and status so you can look it up later by task id.

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

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")
app = Celery("myproject")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

# settings.py
CELERY_BROKER_URL = "redis://localhost:6379/0"     # the queue
CELERY_RESULT_BACKEND = "redis://localhost:6379/1" # store results

# start a worker in a terminal:
# celery -A myproject worker -l info

# The WORKER is a separate process that pulls messages off the broker and
# runs them, storing the return value in the RESULT BACKEND. This simulates
# the full round trip: enqueue -> worker runs -> result stored.

class Broker:
    def __init__(self):
        self.queue = []
    def push(self, message):
        self.queue.append(message)
    def pop(self):
        return self.queue.pop(0) if self.queue else None

class ResultBackend:
    def __init__(self):
        self.results = {}
    def store(self, task_id, value):
        self.results[task_id] = value
    def get(self, task_id):
        return self.results.get(task_id)

broker = Broker()
backend = ResultBackend()

def add(x, y):
    return x + y

# Producer side: enqueue two tasks
broker.push({"id": "t1", "func": add, "args": (2, 3)})
broker.push({"id": "t2", "func": add, "args": (10, 20)})

# Worker side: drain the queue and store results
def run_worker():
    while True:
        msg = broker.pop()
        if msg is None:
            break
        value = msg["func"](*msg["args"])
        backend.store(msg["id"], value)
        print(f"worker ran {msg['id']} -> {value}")

run_worker()
print("result of t1:", backend.get("t1"))
print("result of t2:", backend.get("t2"))

# EXPECTED:
# worker ran t1 -> 5
# worker ran t2 -> 30
# result of t1: 5
# result of t2: 30

# ✅ Expected output:
# worker ran t1 -> 5
# worker ran t2 -> 30
# result of t1: 5
# result of t2: 30

3 Retries and Periodic Tasks with Beat

Background work fails sometimes — an API times out, a payment gateway hiccups. With bind=True and self.retry you re-enqueue a failing task up to max_retries times, optionally after a delay. For work that should run on a schedule, Celery Beat is a scheduler that puts tasks on the broker at fixed times defined in beat_schedule; the regular workers still execute them.

# tasks.py — retry on transient failure
@shared_task(bind=True, max_retries=3, default_retry_delay=10)
def charge_card(self, order_id):
    try:
        gateway.charge(order_id)
    except GatewayTimeout as exc:
        raise self.retry(exc=exc)   # re-enqueue, up to 3 times

# settings.py — run a task every night with Beat
from celery.schedules import crontab
CELERY_BEAT_SCHEDULE = {
    "nightly-report": {
        "task": "myapp.tasks.build_report",
        "schedule": crontab(hour=2, minute=0),   # 2:00 AM daily
    },
}

# start the scheduler alongside the worker:
# celery -A myproject beat -l info

# RETRIES: if a task raises, Celery can re-enqueue it up to max_retries.
# This simulates a flaky task that fails twice then succeeds on the 3rd try.

class TaskRunner:
    def __init__(self, max_retries=3):
        self.max_retries = max_retries

    def run(self, func, *args):
        attempt = 0
        while True:
            try:
                attempt += 1
                result = func(*args, attempt=attempt)
                print(f"attempt {attempt}: success")
                return result
            except Exception as e:
                print(f"attempt {attempt}: failed ({e})")
                if attempt > self.max_retries:
                    print("giving up after max_retries")
                    raise
                # Celery would re-enqueue with a countdown; we just loop

def flaky_charge(card, attempt):
    # fails on the first two attempts, succeeds on the third
    if attempt < 3:
        raise RuntimeError("gateway timeout")
    return f"charged {card}"

runner = TaskRunner(max_retries=3)
result = runner.run(flaky_charge, "****1234")
print("final result:", result)

# EXPECTED:
# attempt 1: failed (gateway timeout)
# attempt 2: failed (gateway timeout)
# attempt 3: success
# final result: charged ****1234

# ✅ Expected output:
# attempt 1: failed (gateway timeout)
# attempt 2: failed (gateway timeout)
# attempt 3: success
# final result: charged ****1234

🎯 Your Turn: Enqueue, Don't Run

Replace the blank with .delay so the task is queued instead of running right now.

# 🎯 YOUR TURN: enqueue a task instead of running it now.
# Replace ___ with .delay so the call is queued, not executed immediately.

queue = []

def shared_task(func):
    func.delay = lambda *a: queue.append((func.__name__, a)) or "queued"
    return func

@shared_task
def resize_image(path):
    return f"resized {path}"

status = resize_image___("photo.jpg")
print("status:", status)
print("queue:", queue)

# EXPECTED OUTPUT:
# status: queued
# queue: [('resize_image', ('photo.jpg',))]

Common Errors (And How to Fix Them)

❌ My task never runs

You enqueued it with .delay() but no worker is running.

✅ Fix: start a worker with celery -A myproject worker -l info.

❌ I passed a whole model object and got a serialization error

Task arguments are serialized to the broker, and full objects don't serialize cleanly.

✅ Fix: pass the primary key (e.g. user.id) and re-fetch inside the task.

❌ I can't get a task's return value

No result backend is configured, so results aren't stored.

✅ Fix: set CELERY_RESULT_BACKEND, then read result.get() by task id.

📋 Quick Reference

🎯 Mini Challenge: Build a Mini Task Queue

Build a small Celery-like system with a task registry, a broker, a worker, and a result backend. Enqueue two tasks with .delay(), run the worker, and fetch the results.

# 🎯 Mini Challenge: build a small in-process Celery-like system with a
# task registry, a broker, a worker, and a result backend. Enqueue two
# tasks with .delay(), run the worker, and fetch the results.

class Celery:
    def __init__(self):
        self.broker = []
        self.results = {}
        self._next_id = 0

    def task(self, func):
        def delay(*args, **kwargs):
            self._next_id += 1
            task_id = f"task-{self._next_id}"
            self.broker.append((task_id, func, args, kwargs))
            return task_id
        func.delay = delay
        return func

    def work(self):
        while self.broker:
            task_id, func, args, kwargs = self.broker.pop(0)
            self.results[task_id] = func(*args, **kwargs)

    def result(self, task_id):
        return self.results.get(task_id)

app = Celery()

@app.task
def add(x, y):
    return x + y

@app.task
def greet(name):
    return f"Hello, {name}!"

# Producer: enqueue (returns immediately, nothing has run)
id1 = add.delay(4, 5)
id2 = greet.delay("Dana")
print("pending in broker:", len(app.broker))

# Worker: drain and run everything
app.work()
print("add result:", app.result(id1))
print("greet result:", app.result(id2))
print("broker now empty:", app.broker == [])

# EXPECTED:
# pending in broker: 2
# add result: 9
# greet result: Hello, Dana!
# broker now empty: True

❓ Frequently Asked Questions

Lesson complete — slow work runs in the background!

You defined tasks with @shared_task, enqueued them with .delay(), traced a task from the broker through a worker to the result backend, and added retries for flaky work plus periodic schedules with Beat.

Practice quiz

Continue this course

• Previous: Real-Time with Django Channels
• Next: Checkpoint: Production Django — Combine a custom-user mixin and a background task — then a quiz
• Quick reference: Django cheat sheet

Frequently asked questions

What is Celery and why use it with Django?

Celery is a distributed task queue that runs work outside the web request, in separate worker processes. You use it with Django to move slow or unreliable jobs — sending email, resizing images, generating reports, calling third-party APIs — off the request so the user gets a fast response. The view enqueues a task and returns immediately, and a worker picks it up and runs it in the background. This keeps your site responsive and lets heavy work scale independently of web traffic.

What is the difference between calling a task normally and calling .delay()?

Calling the function directly, like send_email(addr), runs it right there in the request and blocks until it finishes. Calling send_email.delay(addr) does not run the function at all in that process — it serializes the arguments into a message, puts that message on the broker, and returns an AsyncResult almost instantly. A separate worker later pulls the message and executes the task. So .delay() is the difference between doing the work now and scheduling it to be done in the background.

What are the broker and the result backend?

The broker is the message queue that holds tasks until a worker is ready to run them; Redis and RabbitMQ are the common choices. When you call .delay(), the message goes to the broker, and workers consume from it. The result backend is a separate store that holds the return value and status of each task so you can look them up later by task id. The backend is optional — many tasks are fire-and-forget — but you configure one, often Redis again, when you need to retrieve results.

How do retries and periodic tasks work in Celery?

For retries, you call self.retry inside a task bound with bind=True, and set max_retries; if the task raises a transient error you re-enqueue it, optionally with a countdown delay, until it succeeds or hits the limit. For periodic tasks you use Celery Beat, a scheduler process that puts tasks on the broker on a schedule — for example every night or every five minutes — defined in the beat_schedule setting. Beat only schedules; the normal workers still execute the tasks it enqueues.