Task Queues with Celery and Redis

You just shipped a signup form. A new user signs up, and your request handler fires off a welcome email, generates a PDF receipt, pushes a Slack notification, and resizes a profile picture. That 100 ms endpoint just turned into a 4 second one, and if any of those calls hang, your response hangs with them. The clean fix is a Celery task queue: move the slow work out of the request thread and into a separate worker process that runs on its own clock.

The web request publishes a short message (say, “send the welcome email for user 42”) to a broker like Redis and returns. A separate worker picks it up and runs the work asynchronously. If the worker crashes, the message goes back in the queue. If traffic spikes, you start more workers. The same shape works for a CLI script, a scheduled job, or another service: anything that can talk to Redis can hand work to the worker.

This guide walks through a runnable Celery and Redis setup: project layout, the @app.task decorator, calling tasks with delay and apply_async, retries and backoff, the chain/group/chord canvas primitives, and the Redis-specific gotchas that bite people on day one.

TL;DR

• Use a task queue to move slow, fail-prone work (emails, webhooks, image processing, ETLs) off the request thread.
• Celery has three moving parts: a client (your app), a broker (Redis or RabbitMQ), and one or more worker processes. Workers can run on different machines from the web app.
• @app.task decorates a regular Python function. delay() enqueues; apply_async() adds countdown, eta, queue, and a deterministic task_id for idempotency.
• task_acks_late=True plus idempotent side effects is the price of at-least-once delivery, since tasks can run twice.
• retry_backoff=True with retry_jitter=True retries transient failures without synchronized thundering-herd retries.
• Redis broker gotchas: visibility timeouts cause duplicates on long tasks (idempotency fixes it), ETA jobs live in worker memory, and you should keep the broker and result backend on separate Redis DBs.

Why use a Celery task queue

Threads and multiprocessing both look like solutions until you actually deploy them. Threads share memory, can’t survive a worker process restart, and won’t help you if you need to do a job on a different machine. multiprocessing.Pool is better, but it stays in-process: it dies with your web app, has no persistence, and no built-in retries or scheduling.

A task queue is a separate process (or many of them) that consumes messages from a broker. The publish side is fast: you serialize a small message and push it. The consume side is a long-running process. This gives you:

• At-least-once delivery. Messages survive worker crashes (up to the broker’s visibility timeout).
• Horizontal scaling. Run as many workers as you need, on as many machines as you need.
• Cross-language clients. Anything that can talk to Redis or RabbitMQ can enqueue.
• Retries with backoff. Failed tasks can be rescheduled with jitter to avoid synchronized retries.

Celery is the de facto Python task queue. It ships with the broker plumbing, a worker process, retries, ETA/countdown scheduling, a canvas for composing work into chains and groups, and integrations with Django, Flask, and FastAPI. To make the contrast concrete, here’s the same signup handler written both ways:

# Slow: blocks the request thread on every call
def signup(user_data):
    user = create_user(user_data)
    send_welcome_email(user.email)      # ~2s SMTP round trip
    generate_pdf_receipt(user)          # ~1s PDF generation
    return {"ok": True}

# Fast: enqueues and returns in a few milliseconds
def signup(user_data):
    user = create_user(user_data)
    send_welcome_email.delay(user.id)   # ~1ms to enqueue
    generate_pdf_receipt.delay(user.id)
    return {"ok": True}

The first version does the slow work inline; the second hands it to a worker. The request returns fast either way, but only the second version keeps responding quickly when the email server is slow or the PDF generator is under load.

How Celery works

Three roles, three things in your stack:

1. Client. Your web app or script. It calls task.delay(...) or task.apply_async(...). Celery serializes the arguments and pushes a message to the broker. This is the only part that runs in your request thread, and it should be fast.
2. Broker. Redis, RabbitMQ, or another transport Celery supports via kombu. It holds the queue of pending tasks. For Redis, the queue is a list that the worker pops from.
3. Worker. A separate process started with celery -A myapp.celery_app worker. It pulls messages, imports the task, runs the body, and stores the return value in the result backend if you have one.

You can run one worker locally for development, then scale to N on different machines. The broker is the shared coordination point; the workers are interchangeable.

Project setup

You’ll need Python 3.10+ and a running Redis (any 6.x or 7.x release works). Install Celery with the Redis extras; the bracket syntax pulls the redis client kombu needs for the Redis transport.

pip install "celery[redis]>=5.3"

A minimal but production-shaped layout using a src/-style package. Keeping the Celery app and the task definitions together in a single importable package means the worker can find the task module on startup, your tests can import the same paths your code does, and deployment stays boring:

myapp/
├── pyproject.toml
└── src/
    └── myapp/
        ├── __init__.py
        ├── celery_app.py
        └── tasks.py

The pyproject.toml only needs project metadata and Celery as a dependency. You don’t need poetry, hatch, or a build backend for the examples in this guide; pip install -e . from a basic setup is enough to make the package importable on your machine and on the worker host:

[project]
name = "myapp"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
    "celery[redis]>=5.3",
    "redis>=5.0",
]

The Celery instance lives in celery_app.py. Use different Redis database numbers for the broker and the result backend so the two data sets never collide:

# celery_app.py
from celery import Celery

app = Celery(
    "myapp",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
    include=["myapp.tasks"],
)

app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="UTC",
    enable_utc=True,
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,
    result_expires=3600,
    broker_transport_options={"visibility_timeout": 3600},
)

A few of those options deserve a callout:

• task_serializer="json" and accept_content=["json"] stop anyone from re-introducing pickle (which can deserialize arbitrary code on the worker).
• task_acks_late=True plus task_reject_on_worker_lost=True gives at-least-once delivery: a task is acknowledged only after it runs to completion, and a worker crash returns the message to the queue.
• worker_prefetch_multiplier=1 stops a worker from hoarding messages it hasn’t started yet, which is what you want when tasks vary in duration.

Writing your first task

A task is just a function decorated with @app.task. Celery takes care of serializing the args, finding a worker, running the body, and storing the return value.

# tasks.py
import logging
from myapp.celery_app import app

log = logging.getLogger(__name__)

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

Calling it from anywhere that imports tasks. The web request thread enqueues and returns; a worker process runs the body, so you don’t have to spawn threads or coordinate locks to make this safe under load. Pass IDs and primitives rather than ORM objects so the payload stays JSON-serializable for the broker:

from myapp.tasks import add

result = add.delay(2, 3)
print(result.ready())               # → False (worker hasn't run it yet)
print(result.get(timeout=10))       # → 5

delay() returns an AsyncResult immediately. result.get(timeout=10) blocks the caller for up to 10 seconds, raising celery.exceptions.TimeoutError if the worker doesn’t finish in time. If you don’t want to block, use result.ready() to poll or just fire-and-forget by ignoring the return value.

When bind=True is passed to the decorator, the task instance is injected as self, which gives you access to self.retry(), self.request, and the task logger. You’ll want this for anything that does I/O:

@app.task(bind=True)
def send_welcome_email(self, user_id: int, email: str) -> dict:
    log.info("sending welcome email to %s", email)
    return {"user_id": user_id, "status": "sent"}

Calling tasks: delay vs apply_async

task.delay(*args, **kwargs) is a shortcut for task.apply_async(args, kwargs), so use it for the simple case of kicking off work right now. Reach for apply_async whenever you need to schedule work for the future, route it to a specific worker queue, or set a deterministic task_id that makes retries idempotent:

from datetime import datetime, timezone

send_welcome_email.apply_async(
    args=[42, "user@example.com"],
    countdown=60,         # run in 60 seconds
    expires=300,          # abort if not picked up within 5 minutes
    queue="emails",       # send to a named worker queue
    task_id="welcome-42", # idempotency key — set this to dedupe retries
)

The options you’ll reach for most:

Option	Purpose
countdown	Seconds from now before execution
eta	Absolute UTC datetime of execution
expires	TTL for unacknowledged messages
queue	Send to a named worker queue
task_id	Override the auto-generated UUID; useful for idempotency
link / link_error	Signatures to run on success or failure

eta must be timezone-aware when enable_utc=True is on. With enable_utc=True, naive datetimes are assumed to be UTC; without it, they’re assumed to be local time, which is rarely what you want.

For tests, set app.conf.task_always_eager = True in your test config and .delay() calls run synchronously in the same process. That makes the task body easy to unit test, but retries, ETA, and worker behavior aren’t exercised. Keep a couple of integration tests with a real worker for those paths.

Retries and failure handling

Network blips, SMTP 4xx responses, and database deadlocks are normal. Tasks should treat them as transient.

The decorator form handles the common cases. It declares which exceptions are transient, picks a backoff schedule, and tells the worker to acknowledge only after success, so a crash mid-task returns the message to the queue:

import smtplib
from celery.exceptions import SoftTimeLimitExceeded
from myapp.celery_app import app

@app.task(
    bind=True,
    autoretry_for=(smtplib.SMTPException, ConnectionError, TimeoutError),
    retry_backoff=True,        # exponential: 1, 2, 4, 8, ... seconds
    retry_backoff_max=300,     # cap at 5 minutes
    retry_jitter=True,         # add randomness to avoid synchronized retries
    max_retries=5,
    acks_late=True,
    time_limit=120,            # hard kill in seconds
    soft_time_limit=90,        # raises SoftTimeLimitExceeded after this
)
def send_invoice(self, invoice_id: int) -> dict:
    try:
        # smtp_send(invoice_id)
        return {"invoice_id": invoice_id, "status": "sent"}
    except SoftTimeLimitExceeded:
        self.update_state(state="FAILURE", meta={"reason": "soft_time_limit"})
        raise

For more control, call self.retry(...) from inside the body with custom logic (for example, only retry on specific HTTP status codes).

acks_late=True is the catch. Combined with at-least-once delivery, your task can run twice: once on the original worker, and a second time on another worker if the first died after starting but before acknowledging. Wrap side effects with an idempotency key. A common pattern is a Redis SET key value NX EX 86400 on a key derived from task_id, or a unique constraint on a database row. Tasks that just compute and return a value don’t need anything special; tasks that send emails, charge cards, or call webhooks do.

When retries are exhausted, the task is marked FAILURE and the exception is stored in the result backend. You can listen for that with a task_failure signal handler, or just inspect results through the Flower UI.

Composing work with canvas

Canvas is Celery’s mini-language for wiring tasks together. Three primitives cover most cases:

from celery import chain, group, chord
from myapp.tasks import fetch_orders, charge_order, summarize, add

# Sequential: fetch -> charge -> summarize
chain(fetch_orders.s(user_id=42), charge_order.s(), summarize.s())

# Parallel fan-out: charge 10 orders at once
group(charge_order.s(order_id=i) for i in range(10))

# Group + callback: charge in parallel, then summarize the joined list
chord(
    group(charge_order.s(order_id=i) for i in range(10)),
    summarize.s(),
)

A signature is a frozen call: add.s(2, 2) returns a Signature with x=2, y=2 bound. When the previous task in a chain finishes, its return value is prepended as the first argument of the next task. That’s how chain(fetch_orders.s(...), charge_order.s()) passes the orders list to charge_order.

Chord requires a result backend because the callback receives the joined group result. If you only need fire-and-forget, stick to chain and group.

Redis broker caveats

Redis is fast and easy to run, which is why it’s the default broker for new Celery projects. It isn’t a real message broker though, and a few defaults are worth knowing about before you push to production:

Visibility timeout. When a worker pulls a message from Redis, the message is hidden from other workers for visibility_timeout seconds (default 1 hour). If your task runs longer than that, Redis redelivers it to another worker. Combined with acks_late=True, that means duplicate execution is possible. The fix is idempotent tasks, not a longer timeout.

ETA and countdown are held in worker memory. Tasks scheduled with eta or countdown are immediately fetched by the worker and held in its process until the scheduled time passes. The Celery docs warn about this explicitly. For jobs that need to run hours or days later, use a database-backed scheduler (celerybeat-sqlalchemy, django-celery-beat, or celery-singleton) instead of the in-memory beat.

Separate broker and result backend DBs. The example uses DB 0 for the broker and DB 1 for the result backend. They are different data sets with different access patterns; sharing a DB number is asking for cross-talk.

Result expiry. result_expires defaults to 1 day. If you call result.get() on a task older than that, the result is gone. Set it explicitly to whatever your monitoring needs.

Sentinel for HA. If you want Redis Sentinel failover, point the broker at sentinel://host1:26379;host2:26379 and set broker_transport_options = {"master_name": "cluster1"}.

Common mistakes

A few things that go wrong the first time and are easy to fix once you know. Most of these come from real outages rather than hypothetical edge cases, and a couple of them have cost real money in production:

• Forgetting to start a worker. task.delay() just enqueues. If no worker is consuming, the message sits in Redis until visibility_timeout expires and then redelivers to a worker that still doesn’t exist. Check celery -A myapp.celery_app inspect ping returns pong from each worker.
• Pickle in the serializer. Someone reverts task_serializer to pickle because they want to pass a custom class. Now a malicious or accidental message can execute code on the worker. Keep json, and only pass dicts, lists, strings, numbers, and IDs.
• Non-idempotent side effects with acks_late=True. Two workers, one task, two emails sent. Add the idempotency key the day you add the task, not after the incident.
• Time zones on eta. A naive datetime in your local zone will be misread. Always pass a UTC datetime (or convert from a zoneinfo-aware one first).
• task_always_eager=True leaking into production. A common .env leak from a test config. In production it short-circuits the worker entirely, so .delay() runs in the request thread and you’ve lost the queue. Guard it behind if DEBUG:.
• Web framework can’t find tasks. Flask and FastAPI need @shared_task from celery (or app.autodiscover_tasks(["myapp"]) in the Celery config) so the worker process imports the task module. Otherwise the worker boots fine but can’t resolve the task name and logs NotRegistered.

Frequently asked questions

Is Celery with Redis good enough for production?

Yes for most workloads. Redis is fast, well-understood, and easy to run. The main trade-offs are no built-in high availability beyond Sentinel, and ETA tasks live in worker memory rather than the broker. If you need HA out of the box and durable scheduling, RabbitMQ is a stronger choice. For most web apps, Redis is fine.

How many Celery workers should I run?

Start with about 2 × CPU cores per worker machine, then tune from there. The right number depends on task type: CPU-bound tasks need more processes, I/O-bound tasks can run many more because they spend most of their time waiting on the network. Use --concurrency=N to override the per-worker default, and watch queue depth and worker idle time in the Flower UI.

What happens if a task runs longer than the visibility timeout?

Redis redelivers the message to another worker. With acks_late=True, the original worker may have already started the task, so it can run twice. The fix is idempotent tasks: derive a key from task_id and use SET key value NX EX 86400 in Redis or a unique constraint in the database. The stdlib queue.Queue alternative doesn’t have this problem because it runs in one process.

When to pick Celery over RabbitMQ

Pick Celery with Redis for most web apps: it’s fast, easy to run, and well-understood. Celery and RabbitMQ aren’t alternatives at the same layer; Celery is the client/worker framework and RabbitMQ is one of the brokers it can use. When you need durable scheduled tasks, complex routing rules, or stronger delivery guarantees out of the box, point Celery at a RabbitMQ broker instead of Redis and most of your task code stays the same. A few options like priority are honored only on RabbitMQ.

Does Celery work with asyncio?

Yes, since Celery 5.0. Async tasks use the @app.task decorator with async def bodies, and Celery runs them on an event loop. The canvas primitives (chain, group, chord) work the same way. For short, in-process concurrency inside a single request, asyncio.TaskGroup is lighter; Celery is for work that needs to survive a request and run on a separate process.

Conclusion

A Celery task queue is the right tool when work is slow, fail-prone, or both. The setup is small: a celery_app.py with a broker URL, a few app.conf.update(...) lines, an @app.task decorator, and a worker process. The operational pitfalls (visibility timeouts, ETA held in memory, idempotency under acks_late) are all avoidable once you know they exist. Task queues buy you the most headroom for the least code, and the move from a synchronous request thread to background work is the one refactor that pays off the fastest.

From here, the natural next step is wiring Celery into a web app. The FastAPI quickstart shows a concrete integration, and the email automation tutorial walks through a more elaborate end-to-end use case.

See also

• queue module: the in-process queue.Queue alternative for single-process work.
• logging module: Celery workers use stdlib logging; configure it before the first deploy.
• smtplib module: the standard library way to send the emails the example tasks produce.
• Python decorators: what’s actually happening when you write @app.task.
• Async task groups: asyncio.TaskGroup solves a different problem; use it for short, in-process concurrency, not long-running background work.