Home / MAD2 / Week 9

Week 9 · Asynchronous jobs: Redis and Celery

The one-minute story

By the end of this week you can:

• explain synchronous versus asynchronous work and why slow jobs must leave the request;
• explain the roles of broker, worker and result backend, and what Redis does in this setup;
• write a Celery task, call it with .delay() and .apply_async(), and predict what it returns;
• configure Celery with Flask (app context) and start a worker and a beat scheduler;
• design a CSV export flow (trigger, background build, status polling, download) and a daily reminder;
• explain retries, idempotency and failure handling.

Words you need

Idea 1: Synchronous versus asynchronous, and why jobs leave the request

A Flask worker handles one request at a time (per worker process). A route that builds a PDF and sends an email takes, say, 12 seconds. During that time that worker is busy: other users queue up and the page feels dead, and the browser may time out. Using Celery does not make the job faster (it does not add CPU cores); it moves the job to another process so the web request can return in milliseconds.

# synchronous: the user waits 12 seconds, the server is blocked
@app.get("/report")
def report():
    pdf = build_pdf()          # 10 s
    send_email(pdf)            # 2 s
    return {"done": True}

# asynchronous: respond at once, work happens elsewhere
@app.get("/report")
def report():
    task = build_and_send.delay()
    return {"task_id": task.id}, 202      # 202 Accepted: work has been queued

Good fits for Celery: sending emails, generating reports and CSV/PDF files, image or video processing, calling slow external APIs, nightly cleanups, periodic emails or reminders. Bad fits: handling real-time user interactions, generating the page the user is waiting for, serving live notifications (that is for WebSockets or Server-Sent Events).

Do not use time.sleep() as a trick to wait for a job inside a route: it blocks the worker, which is the very problem we are solving.

Idea 2: Message broker concept

Think of 50 servers that must talk to each other. Point-to-point: every pair needs its own connection, so the number of links grows as O(n squared). With a central broker: each server connects only to the broker, so links grow as O(n). This counting question appears again and again; remember: point-to-point O(n^2), broker O(n).

Benefits of a broker: it decouples producers and consumers (they need not run at the same time, or know each other), makes the system scalable (add more workers), absorbs traffic spikes (messages wait in the queue until processed) and enables batch processing. If the consumer is slower than the producer, messages accumulate in the queue (they are not dropped, the producer does not block). A topic is a named channel to which messages are published. A broker does not let servers talk "directly without an intermediary", that is the opposite.

Two delivery styles: push queues (the broker pushes to consumers, good for real-time) and pull queues (consumers fetch when ready, good for batch work). Celery workers pull.

Idea 3: Redis, the board in the kitchen

It is more than strings. Typical data types and commands (I checked the behaviour with a Redis emulation):

127.0.0.1:6379> SET name Asha
OK
127.0.0.1:6379> EXPIRE name 60
(integer) 1
127.0.0.1:6379> TTL name
(integer) 60            # seconds left; -1 = no expiry, -2 = key does not exist
127.0.0.1:6379> LPUSH q a b
(integer) 2
127.0.0.1:6379> LRANGE q 0 -1
1) "b"
2) "a"

LPUSH pushes to the left, so the last pushed (b) is first. A key with EXPIRE disappears by itself, which is how caches clean up. redis-cli opens the interactive shell; PING should answer PONG.

Persistence. Because data lives in memory, a restart would lose it unless persistence is on: RDB takes periodic snapshots to a file; AOF (append-only file) logs every write. Redis can use both. For Celery as a queue and cache, losing some data on restart is often acceptable; for your only copy of important data, keep it in the SQL database.

Python client: r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True), then r.set("k","v"), r.get("k"). Redis has numbered databases 0..15 inside one server; the URL redis://localhost:6379/1 means database 1. Redis also does not have to be the broker: Celery can use RabbitMQ and others, and a system can have several brokers.

Redis on Windows

Redis is built for Linux. On Windows: use WSL (install Ubuntu, then sudo apt install redis-server, sudo service redis-server start) or Docker (docker run -p 6379:6379 redis). Port 6379 is then reachable as localhost. Check with redis-cli ping.

Idea 4: Celery architecture and your first task

 Flask app           Redis (broker)         Celery worker(s)        Redis (backend)
 (producer)  --1--> [ task queue ] --2--> run task function --3--> [ state + result ]
     |                                                                 ^
     +---------------------- 4. AsyncResult(id).state / .get() ---------+

In the real paper the code Celery('tasks', broker='redis://localhost:6379/1', backend='redis://localhost:6379/2') is explained like this: the broker stores task requests until a worker picks and executes them; the backend stores the result after execution. They can use the same Redis server in different databases; nothing forces different ports. The broker does not "track which worker processes which task".

# tasks.py
from celery import Celery
import time

celery = Celery("tasks",
                broker="redis://localhost:6379/0",
                backend="redis://localhost:6379/1")

@celery.task
def long_task():
    time.sleep(10)
    return "Task completed"

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

Start the worker (in its own terminal, from the folder with tasks.py), then call from Python or Flask:

celery -A tasks worker -l info          # for a Flask app: celery -A app.celery worker -l info

result = add.delay(4, 6)
print(result)            # a task id string such as 9c1f...: it is an AsyncResult, NOT 10
print(type(result))      # <class 'celery.result.AsyncResult'>
print(result.status)     # 'PENDING' right away (worker not finished or not started)
print(result.get(timeout=5))   # 10, waits up to 5 s for the worker
print(result.status)     # 'SUCCESS'

check = add.delay(4, 6)
print(check.status)              # (a)
print(check.get(timeout=5))      # (b)

delay versus apply_async

task.delay(a, b) is a shortcut for task.apply_async(args=(a, b)). Use apply_async for options:

send_email.delay("a@x.com")                                  # send now
send_email.apply_async(args=["a@x.com"], countdown=10)       # run 10 s later
send_email.apply_async(args=["a@x.com"], eta=datetime_utc)   # run at a time
send_email.apply_async(args=["a@x.com"], expires=60)         # drop if not started in 60 s

With countdown=10 the message is sent to the queue immediately but the worker starts it only after the 10 seconds have passed (it does not wait 10 seconds before dispatching). .delay() dispatches immediately and does not wait for any user input.

Task states

Idea 5: Celery with Flask, and the app context

The worker is a separate process. Your task often needs the database or current_app, which only exist inside a Flask app context. The standard pattern (from the Flask docs) makes every task run inside one:

# celery_app.py
from celery import Celery, Task
from flask import Flask

def celery_init_app(app: Flask) -> Celery:
    class FlaskTask(Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery = Celery(app.name, task_cls=FlaskTask)
    celery.config_from_object(app.config["CELERY"])
    celery.set_default()
    app.extensions["celery"] = celery
    return celery

# app.py
app = Flask(__name__)
app.config["CELERY"] = dict(
    broker_url="redis://localhost:6379/0",
    result_backend="redis://localhost:6379/1",
    timezone="Asia/Kolkata",
)
celery = celery_init_app(app)

I ran this pattern with eager mode: a task that returns current_app.name returned the app's name, proving the context works. Then start the worker with celery -A app.celery worker -l info (module app, the variable celery). Some course templates instead create the Celery object and then set celery.conf.update(...): check your course notes for the exact style. Rules:

• The worker must be able to import your tasks (include=["tasks"] or put them in the module you pass to -A).
• Restart the worker after you change task code; it loads code once.
• Arguments and return values travel as JSON, so pass simple things: ids, strings, numbers, not SQLAlchemy objects (job_report.delay(user) is bad; pass user.id and load the user inside the task).
• On Windows the default pool does not work well; start the worker with celery -A app.celery worker -l info -P solo. WSL or Docker avoids this.

Idea 6: Scheduled (beat) versus on-demand tasks

Celery beat is only a scheduler. At the due time it puts a message in the broker and stops. A worker must be running to pick it up. Without a worker the morning digest is never sent. (This exact claim appeared as a question: "the worker is optional, beat handles execution" is false.) You start beat separately: celery -A app.celery beat -l info. Do not schedule with @app.task_periodic (no such thing); a fixed-interval email needs Celery beat, no external scheduler.

from celery.schedules import crontab

celery.conf.beat_schedule = {
    "daily-reminder": {
        "task": "tasks.send_reminders",
        "schedule": crontab(hour=18, minute=0),               # every day 18:00
    },
    "monthly-report": {
        "task": "tasks.monthly_report",
        "schedule": crontab(minute=0, hour=8, day_of_month=1),  # 1st of month, 08:00
    },
    "every-10-min": {
        "task": "tasks.send_email",
        "schedule": 600.0,                                    # seconds
        "args": ("admin@x.com",),
    },
}

I printed these schedules: crontab(hour=18, minute=30) shows as 30 18 * * *; crontab(minute='*/10') is every 10 minutes; crontab(hour=7, minute=0, day_of_week='mon') is Mondays at 07:00. The task name is the module path plus function name (tasks.send_reminders). Set timezone so "18:00" means India time.

Idea 7: The async CSV export flow

Problem: exporting all of a user's data takes 20 to 40 seconds. A fetch('/static/data.csv') after a fixed setTimeout of 41 seconds is a bad design: the job may take longer, and the user waits blindly. The right design is trigger, build, poll, download:

1. Browser calls POST /api/export. Flask starts the task and returns 202 with a task id.
2. The task builds the file in the background and returns its filename.
3. Browser short-polls GET /api/export/<task_id> every few seconds until state is SUCCESS.
4. Browser then fetches the download link.

# tasks.py
import csv, os
from celery_app import celery
from models import Item

@celery.task
def export_items(user_id):
    path = f"exports/items_{user_id}.csv"
    os.makedirs("exports", exist_ok=True)
    with open(path, "w", newline="") as f:
        w = csv.writer(f)
        w.writerow(["id", "name"])
        for it in Item.query.filter_by(user_id=user_id):
            w.writerow([it.id, it.name])
    return path

# routes
from celery.result import AsyncResult

@app.post("/api/export")
@jwt_required()
def start_export():
    task = export_items.delay(int(get_jwt_identity()))
    return {"task_id": task.id}, 202

@app.get("/api/export/<task_id>")
@jwt_required()
def export_status(task_id):
    res = AsyncResult(task_id, app=celery)
    if res.state == "SUCCESS":
        return {"state": "SUCCESS", "download": f"/api/download/{task_id}"}
    if res.state == "FAILURE":
        return {"state": "FAILURE"}, 500
    return {"state": res.state}          # PENDING / STARTED / RETRY

// Vue method: short polling
async startExport() {
  const r = await fetch('/api/export', { method: 'POST', headers: this.authHeader })
  const { task_id } = await r.json()
  const timer = setInterval(async () => {
    const s = await (await fetch('/api/export/' + task_id, { headers: this.authHeader })).json()
    if (s.state === 'SUCCESS') { clearInterval(timer); this.downloadUrl = s.download }
    if (s.state === 'FAILURE') { clearInterval(timer); this.error = 'Export failed' }
  }, 3000)
}

Short polling = asking on a fixed timer; it is the way to learn the state of an async task and trigger an action when done. Long polling = the server holds the request open until something happens, giving near real-time updates over plain HTTP (so "long polling cannot be done over HTTP" is false). A webhook is not short polling: the server pushes a message to a URL when an event happens. Server-Sent Events (SSE) keep one connection open for the server to push to the browser (one-way; Flask-SSE uses Redis as a shared channel).

Idea 8: Retries, idempotency, failure handling, monitoring

@celery.task(bind=True, autoretry_for=(ConnectionError,), retry_backoff=True, max_retries=3)
def send_mail(self, to):
    smtp_send(to)

# manual style
@celery.task(bind=True, max_retries=3)
def send_mail2(self, to):
    try:
        smtp_send(to)
    except ConnectionError as exc:
        raise self.retry(exc=exc, countdown=30)    # try again in 30 s

• Retries: transient failures (network, SMTP) are retried a limited number of times. After max_retries the state becomes FAILURE.
• Idempotency: a message may be delivered twice (worker crash, retry). Design tasks so repeating them is harmless: check "already sent?" before sending, use INSERT ... ON CONFLICT or unique keys, write output to a file named by task or user id so a rerun overwrites the same file instead of duplicating.
• Failure handling: log exceptions, set a FAILURE state, show "Export failed, try again" in the UI; never let the polling loop run forever (set a maximum number of polls).
• Pass ids not objects, and re-read the database inside the task so you work on fresh data.
• Monitoring: Flower is a web dashboard for Celery (celery -A app.celery flower) showing workers, tasks and states. celery -A app.celery inspect active lists running tasks.
• Queue growth: if tasks arrive faster than workers finish, messages accumulate in the queue; add workers (-c 4 concurrency, or more worker machines).

celery.conf.beat_schedule = {
    "weekly-digest": {
        "task": "tasks.weekly_digest",
        "schedule": crontab(hour=7, minute=30, day_of_week="mon"),
    }
}

Exam radar

Quick quiz

Viva: say it out loud

Recap

Practise with real exam questions

These come straight from past papers (with the real figures and the official answer key). Open 60 real questions on this week   All End Term papers