Background Tasks (Celery & RQ)

Reviewed & published by Brayan K

A background task moves slow work off the request thread, so your route returns instantly while a separate worker runs the heavy job and stores its result.

Part of the free Flask 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 build an in-process job queue, mimic Celery's .delay() and result backend, and see the real Celery and RQ code you'd run in production.

What You'll Learn in This Lesson

• Why slow work belongs off the request thread
• Enqueueing jobs and processing them with a worker
• How a broker and result backend fit together
• Mimicking Celery's .delay() in plain Python
• Defining a real Celery task with a broker
• When to reach for RQ instead of Celery

1 Why Offload? An In-Process Job Queue

When a user signs up, you might send a welcome email. Email is slow — an SMTP round-trip can take seconds. If you do it inside the request, the user waits and a worker is tied up. The fix: enqueue the slow job and return immediately, then let a separate worker run it.

The runnable example models this with a deque as the queue and a worker() that drains it. The route enqueues the email and responds accepted right away; the jobs only run when the worker is called.

# app.py — an in-process job queue: enqueue now, process later
import time
from collections import deque
from flask import Flask, jsonify

app = Flask(__name__)

job_queue = deque()    # pending jobs
results = {}           # job_id -> result

def send_email(to, subject):
    # pretend this is slow (an SMTP round-trip)
    time.sleep(0)  # no real delay needed for the demo
    return f"sent '{subject}' to {to}"

@app.route("/signup/<user>")
def signup(user):
    # enqueue the slow work instead of doing it in the request
    job_id = len(results) + len(job_queue) + 1
    job_queue.append((job_id, send_email, (f"{user}@example.com", "Welcome!")))
    return jsonify(status="accepted", job_id=job_id)   # 202-style: queued

def worker():
    # a background worker drains the queue
    while job_queue:
        job_id, func, args = job_queue.popleft()
        results[job_id] = func(*args)

# The request returns instantly; the worker runs the jobs separately:
with app.test_client() as c:
    print(c.get("/signup/ada").get_json())   # {'job_id': 1, 'status': 'accepted'}
    print(c.get("/signup/bob").get_json())   # {'job_id': 2, 'status': 'accepted'}
    print("pending jobs:", len(job_queue))   # pending jobs: 2
    worker()                                 # process the queue
    print(results)                           # {1: "...ada...", 2: "...bob..."}

# app.run() would serve this forever, so it goes last and stays commented
# out — the test client above is how Flask itself exercises routes:
#     if __name__ == "__main__":
#         app.run(debug=True)

Both signups return accepted immediately while two jobs sit pending. Only after worker() runs do the results appear — exactly how Celery and RQ separate enqueueing from execution.

2 .delay() and the Result Backend

Celery decorates a function so you can call task.delay(args) to enqueue it. That call returns immediately with a task id; the function does not run in your web process. The worker later writes the return value to a result backend you can poll by id.

The runnable example below recreates that shape: a @task decorator that adds a .delay() method, a queue, and a result_backend dict tracking each task's state from PENDING to SUCCESS.

# .delay() style enqueue + a result backend, in plain Python
queue = []
result_backend = {}   # task_id -> {"state": ..., "result": ...}
_next_id = [0]

def task(fn):
    # mimic Celery's @task: gives the function a .delay() method
    def delay(*args, **kwargs):
        _next_id[0] += 1
        tid = _next_id[0]
        result_backend[tid] = {"state": "PENDING", "result": None}
        queue.append((tid, fn, args, kwargs))   # enqueue, return immediately
        return tid
    fn.delay = delay
    return fn

@task
def add(a, b):
    return a + b

def run_worker():
    while queue:
        tid, fn, args, kwargs = queue.pop(0)
        result_backend[tid] = {"state": "SUCCESS", "result": fn(*args, **kwargs)}

# Caller enqueues and gets a task id back instantly:
tid = add.delay(2, 40)
print("queued task:", tid, result_backend[tid])   # queued task: 1 {'state': 'PENDING', 'result': None}
run_worker()                                       # the worker processes it
print("finished:", result_backend[tid])            # finished: {'state': 'SUCCESS', 'result': 42}

# ✅ Expected output:
# queued task: 1 {'state': 'PENDING', 'result': None}
# finished: {'state': 'SUCCESS', 'result': 42}

3 RQ: the Simpler Alternative

RQ (Redis Queue) trades Celery's many features for a tiny, readable API. You create a Queue backed by Redis and call queue.enqueue(func, args). It is ideal when you just need "run this function later" without scheduling, routing, or multiple brokers.

The runnable example keeps mirroring the concept — enqueue a function reference plus its arguments, then a worker pops and runs it — so you can see RQ's mental model before reading its real code.

# RQ's mental model: queue.enqueue(func, *args) then a worker runs it
queue = []                       # stands in for an RQ Queue (backed by Redis)
finished = {}

def enqueue(func, *args):
    job_id = f"job-{len(queue) + len(finished) + 1}"
    queue.append((job_id, func, args))
    return job_id                # RQ returns a Job whose .id you can track

def resize_image(name, width):
    return f"resized {name} to {width}px"

def run_worker():
    while queue:
        job_id, func, args = queue.pop(0)
        finished[job_id] = func(*args)

jid = enqueue(resize_image, "avatar.png", 128)
print("enqueued:", jid, "| pending:", len(queue))   # enqueued: job-1 | pending: 1
run_worker()
print("result:", finished[jid])                     # result: resized avatar.png to 128px

# ✅ Expected output:
# enqueued: job-1 | pending: 1
# result: resized avatar.png to 128px

🔬 Celery vs RQ Cheat Sheet

🎯 Your Turn: Fill in the Blanks

Complete the queue below. Replace each ___ so a job is enqueued and a worker runs it.

# 🎯 YOUR TURN: replace each ___ to finish the queue
queue = []
results = {}

def enqueue(fn, *args):
    job_id = len(results) + len(queue) + 1
    queue.___((job_id, fn, args))    # add the job to the queue
    return job_id

def slow_square(n):
    return n * n

def worker():
    while queue:
        job_id, fn, args = queue.pop(0)
        results[job_id] = fn(___)    # call fn with its stored args

jid = enqueue(slow_square, 7)
print("queued:", jid)            # queued: 1
worker()
print("result:", results[jid])   # result: 49

Common Errors & Quick Reference

❌ Task runs in the request, blocking the response

You called the function directly instead of task.delay(...). Use .delay() (Celery) or q.enqueue(...) (RQ) so it goes on the broker and a worker runs it.

❌ "No result backend configured"

You tried to read res.result without setting backend=. A broker queues tasks; a separate result backend stores return values. Configure both if you need results.

🏆 Mini Challenge: A Retrying Queue

Real task queues retry failed jobs. Build a worker that does too.

• Each job tracks tries and a max retry count
• On failure, re-enqueue the job until max is reached
• A job that fails once but succeeds on retry ends as ok

# 🏆 Mini Challenge: a retrying job queue
queue = []
results = {}

def flaky_double(n, _state={"calls": 0}):
    _state["calls"] += 1
    if _state["calls"] == 1:        # fail the first attempt
        raise ValueError("transient error")
    return n * 2

def worker():
    while queue:
        job = queue.pop(0)
        job["tries"] += 1
        try:
            results[id(job)] = ("ok", job["fn"](*job["args"]))
        except Exception as e:
            if job["tries"] < job["max"]:
                queue.append(job)    # re-enqueue for a retry
            else:
                results[id(job)] = ("failed", str(e))

job = {"fn": flaky_double, "args": (21,), "tries": 0, "max": 3}
queue.append(job)
worker()
print(results[id(job)])               # ('ok', 42)
print("final tries:", job["tries"])   # final tries: 2

❓ Frequently Asked Questions

Lesson complete — your slow work runs off the request thread!

You built an in-process job queue, mimicked Celery's .delay() and a result backend, met RQ's simpler API, and saw the real production code for both.

Practice quiz

Continue this course

• Previous: Rate Limiting (Flask-Limiter)
• Next: Real-Time with WebSockets (Flask-SocketIO) — Push live updates with WebSocket events, rooms, and broadcasts
• Quick reference: Flask cheat sheet

Frequently asked questions

Why move work to a background task?

A web request should return fast. Slow work — sending email, resizing images, calling a third-party API, generating a report — blocks the response and ties up a worker. Offloading it to a background task lets the request return immediately (often with a 202 Accepted) while the heavy work runs elsewhere.

What is a broker and a result backend?

A broker is a message queue (commonly Redis or RabbitMQ) that holds enqueued tasks until a worker picks them up. A result backend is separate storage (often Redis or a database) where the worker writes a task's return value so the web app can fetch it later by task id.

What is the difference between Celery and RQ?

Both run Python functions in the background. Celery is feature-rich — multiple brokers, scheduling (beat), retries, routing, workflows — but heavier to configure. RQ (Redis Queue) is simpler: it only uses Redis and has a tiny API, which makes it a great choice for straightforward job queues.

What does .delay() do in Celery?

calling task.delay(args) serializes the task name and arguments, pushes them onto the broker, and returns an AsyncResult immediately — it does not run the function in your web process. A separate Celery worker pops the message and executes the task. .apply_async() is the fuller version with extra options.

How do I get a task's result later?

Celery returns an AsyncResult whose .id you can store. Later you reload it with celery.AsyncResult(task_id) and read .state (PENDING, STARTED, SUCCESS, FAILURE) and .result. The value lives in the result backend you configured.