Celery Cheatsheet

Celery is a distributed task queue for running work outside the request/response path. A producer sends a task message to a broker, workers consume messages, and an optional result backend stores task state and return values.

Core Concepts

Term	Meaning
Producer	Your web app, CLI, cron job, or service that sends tasks
Broker	Message transport such as Redis or RabbitMQ
Worker	Process that executes tasks from queues
Queue	Named message stream workers consume from
Task	Python callable registered with Celery
Result backend	Storage for task state/results such as Redis, database, or RPC
Beat	Scheduler that sends periodic tasks
Canvas	Primitives for workflows: chains, groups, chords, maps

Web/API process -> broker -> worker process -> optional result backend

Installation

# Base install
pip install celery

# Redis broker/backend support
pip install "celery[redis]"

# RabbitMQ uses AMQP support included through Kombu dependencies
pip install celery

# Common extras
pip install flower              # web monitoring UI
pip install django-celery-beat  # DB-backed periodic tasks for Django
pip install django-celery-results

Minimal App

Create proj/celery_app.py:

from celery import Celery

app = Celery(
    "proj",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
)

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

Run a worker:

celery -A proj.celery_app worker --loglevel=INFO

Call a task:

from proj.celery_app import add

result = add.delay(2, 3)
print(result.id)
print(result.get(timeout=10))

Project Layout

proj/
  __init__.py
  celery_app.py
  tasks.py

proj/celery_app.py:

from celery import Celery

app = Celery("proj")
app.config_from_object("proj.celeryconfig")
app.autodiscover_tasks(["proj"])

proj/tasks.py:

from .celery_app import app

@app.task(name="proj.tasks.send_email")
def send_email(user_id):
    ...

proj/celeryconfig.py:

broker_url = "redis://localhost:6379/0"
result_backend = "redis://localhost:6379/1"
timezone = "UTC"
task_serializer = "json"
accept_content = ["json"]
result_serializer = "json"

Command Line

Workers

# Start worker
celery -A proj.celery_app worker --loglevel=INFO

# Set concurrency
celery -A proj.celery_app worker --concurrency=4

# Consume specific queues
celery -A proj.celery_app worker -Q default,email,high_priority

# Name a worker
celery -A proj.celery_app worker -n worker1@%h

# Enable task events for monitoring
celery -A proj.celery_app worker -E

# Use a pool
celery -A proj.celery_app worker --pool=prefork
celery -A proj.celery_app worker --pool=threads --concurrency=20
celery -A proj.celery_app worker --pool=solo

Beat Scheduler

# Start scheduler
celery -A proj.celery_app beat --loglevel=INFO

# Development only: worker + beat in one process
celery -A proj.celery_app worker -B --loglevel=INFO

Inspect and Control

# Check live workers
celery -A proj.celery_app status

# Registered tasks
celery -A proj.celery_app inspect registered

# Active, reserved, scheduled tasks
celery -A proj.celery_app inspect active
celery -A proj.celery_app inspect reserved
celery -A proj.celery_app inspect scheduled

# Worker stats
celery -A proj.celery_app inspect stats

# Revoke a task
celery -A proj.celery_app control revoke <task-id>

# Revoke and terminate running task
celery -A proj.celery_app control revoke <task-id> --terminate

# Shut down workers
celery -A proj.celery_app control shutdown

Defining Tasks

Basic Task

@app.task
def resize_image(image_id):
    ...

Named Task

@app.task(name="images.resize")
def resize_image(image_id):
    ...

Bound Task

Use bind=True when the task needs self.request, retries, logging context, or custom state.

@app.task(bind=True)
def process_order(self, order_id):
    print(self.request.id)
    print(self.request.retries)

Task Options

@app.task(
    name="reports.generate",
    bind=True,
    max_retries=3,
    default_retry_delay=60,
    soft_time_limit=300,
    time_limit=360,
    ignore_result=False,
)
def generate_report(self, report_id):
    ...

Option	Purpose
`name`	Stable task name used on the broker
`bind=True`	Pass task instance as first argument
`max_retries`	Maximum retry attempts
`default_retry_delay`	Delay between retries in seconds
`autoretry_for`	Exception tuple that triggers automatic retry
`retry_backoff`	Exponential retry delay
`retry_jitter`	Randomize retry delay to reduce spikes
`acks_late`	Acknowledge only after task finishes
`ignore_result`	Do not store result
`soft_time_limit`	Raise exception inside task after limit
`time_limit`	Kill task after hard limit
`rate_limit`	Limit execution rate, such as `10/m`

Calling Tasks

# Shortcut: send immediately
task.delay(arg1, arg2, key="value")

# Full API
task.apply_async(args=[arg1, arg2], kwargs={"key": "value"})

# Delay execution
task.apply_async(args=[42], countdown=60)

# Run at exact time
task.apply_async(args=[42], eta=datetime.utcnow() + timedelta(minutes=10))

# Expire if not executed in time
task.apply_async(args=[42], expires=300)

# Send to specific queue
task.apply_async(args=[42], queue="high_priority")

# Set priority when broker supports it
task.apply_async(args=[42], priority=9)

# Custom task id
task.apply_async(args=[42], task_id="order-123")

Results

res = add.delay(2, 3)

res.id
res.status       # PENDING, STARTED, SUCCESS, FAILURE, RETRY, REVOKED
res.ready()
res.successful()
res.failed()
res.get(timeout=10)
res.result       # return value or exception
res.traceback

Common result methods:

Method	Description
`get(timeout=n)`	Wait for result
`ready()`	Task finished or failed
`successful()`	Task completed successfully
`failed()`	Task failed
`forget()`	Delete stored result
`revoke()`	Ask workers not to execute task

Avoid waiting for Celery results inside web requests unless the wait is short and intentional. It often defeats the point of using a queue.

Retries

Manual Retry

@app.task(bind=True, max_retries=5, default_retry_delay=30)
def fetch_url(self, url):
    try:
        return requests.get(url, timeout=10).text
    except requests.RequestException as exc:
        raise self.retry(exc=exc)

Automatic Retry

@app.task(
    autoretry_for=(requests.RequestException,),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
    max_retries=5,
)
def fetch_url(url):
    return requests.get(url, timeout=10).text

Retry with Custom Countdown

@app.task(bind=True)
def sync_customer(self, customer_id):
    try:
        ...
    except RateLimited as exc:
        raise self.retry(exc=exc, countdown=120)

Idempotency and Reliability

Tasks should be safe to run more than once. Workers can crash after side effects, brokers can redeliver messages, and retries can duplicate work.

@app.task(bind=True, acks_late=True)
def charge_invoice(self, invoice_id):
    invoice = Invoice.objects.get(id=invoice_id)

    if invoice.status == "paid":
        return "already-paid"

    # Use an idempotency key with the external API.
    payment.charge(
        invoice.customer_id,
        amount=invoice.total,
        idempotency_key=f"invoice:{invoice.id}",
    )

    invoice.status = "paid"
    invoice.save(update_fields=["status"])

Production reliability settings:

task_acks_late = True
task_reject_on_worker_lost = True
worker_prefetch_multiplier = 1
task_track_started = True

Setting	Why it matters
`task_acks_late=True`	Redeliver task if worker dies before finishing
`task_reject_on_worker_lost=True`	Reject instead of ack when worker process disappears
`worker_prefetch_multiplier=1`	Prevent one worker from reserving too many tasks
`task_track_started=True`	Report `STARTED` state

Time Limits

@app.task(soft_time_limit=60, time_limit=75)
def import_file(file_id):
    ...

from celery.exceptions import SoftTimeLimitExceeded

@app.task(soft_time_limit=60, time_limit=75)
def import_file(file_id):
    try:
        ...
    except SoftTimeLimitExceeded:
        cleanup_temp_files(file_id)
        raise

Limit	Behavior
`soft_time_limit`	Raises `SoftTimeLimitExceeded` inside the task
`time_limit`	Terminates the worker child process

Routing and Queues

Simple Routing

task_default_queue = "default"

task_routes = {
    "emails.*": {"queue": "email"},
    "reports.*": {"queue": "reports"},
    "images.resize": {"queue": "media"},
}

Start workers for specific queues:

celery -A proj.celery_app worker -Q default
celery -A proj.celery_app worker -Q email,reports
celery -A proj.celery_app worker -Q media --concurrency=2

Send directly:

send_email.apply_async(args=[user_id], queue="email")

Advanced Queue Declaration

from kombu import Exchange, Queue

task_queues = (
    Queue("default", Exchange("default"), routing_key="default"),
    Queue("email", Exchange("email"), routing_key="email.#"),
    Queue("reports", Exchange("reports"), routing_key="reports.#"),
)

task_default_exchange = "default"
task_default_routing_key = "default"

Periodic Tasks

Static Beat Schedule

from celery.schedules import crontab

beat_schedule = {
    "cleanup-every-hour": {
        "task": "maintenance.cleanup",
        "schedule": 3600.0,
        "args": (),
    },
    "send-daily-digest": {
        "task": "emails.daily_digest",
        "schedule": crontab(hour=8, minute=0),
    },
    "sync-every-weekday": {
        "task": "sync.customers",
        "schedule": crontab(minute=0, hour="*/2", day_of_week="mon-fri"),
    },
}

Run beat:

celery -A proj.celery_app beat --loglevel=INFO

Run workers separately:

celery -A proj.celery_app worker --loglevel=INFO

Dynamic Schedules in Django

pip install django-celery-beat
python manage.py migrate

INSTALLED_APPS = [
    ...
    "django_celery_beat",
]

celery -A config beat -S django --loglevel=INFO

Canvas Workflows

Canvas lets you compose tasks into workflows.

Signatures

sig = add.s(2, 3)
sig.delay()
sig.apply_async()

Chain

Run tasks one after another. The previous result is passed to the next task.

from celery import chain

workflow = chain(add.s(2, 3), multiply.s(10), store_result.s())
workflow.delay()

# Pipe syntax
(add.s(2, 3) | multiply.s(10) | store_result.s()).delay()

Group

Run tasks in parallel.

from celery import group

job = group(fetch_url.s(url) for url in urls)
result = job.delay()
values = result.get(timeout=30)

Chord

Run a callback after a group finishes.

from celery import chord

workflow = chord(
    group(fetch_url.s(url) for url in urls),
    summarize_results.s(),
)
workflow.delay()

Immutable Signatures

Use .si() when you do not want previous results injected.

(create_user.s(email) | send_welcome_email.si(email)).delay()

Worker Pools

Pool	Use case
`prefork`	Default for CPU-bound or normal Python tasks
`threads`	I/O-heavy tasks when libraries are thread-safe
`solo`	Debugging, Windows development, simple local runs
`eventlet`	Greenlet-based I/O with monkey patching
`gevent`	Greenlet-based I/O with monkey patching

celery -A proj.celery_app worker --pool=prefork --concurrency=4
celery -A proj.celery_app worker --pool=threads --concurrency=20
celery -A proj.celery_app worker --pool=solo

General guidance:

Workload	Recommendation
CPU-heavy	`prefork`, fewer processes, separate queue
Network I/O	`threads`, `gevent`, or `eventlet` after testing
Long-running tasks	Dedicated queue and `worker_prefetch_multiplier=1`
Memory leaks	Use `--max-tasks-per-child` or `--max-memory-per-child`

Concurrency and Prefetch

# Four concurrent child processes or threads
celery -A proj.celery_app worker --concurrency=4

# Restart child process after N tasks
celery -A proj.celery_app worker --max-tasks-per-child=1000

# Restart child process after memory limit in KB
celery -A proj.celery_app worker --max-memory-per-child=300000

# Reserve one task per worker slot
worker_prefetch_multiplier = 1

# Useful for long-running tasks
task_acks_late = True

Prefetch behavior:

Setting	Effect
`worker_prefetch_multiplier=4`	Each worker process may reserve 4 tasks
`worker_prefetch_multiplier=1`	Fairer distribution for long tasks
`worker_prefetch_multiplier=0`	Worker may consume as many messages as possible

Monitoring

Flower

celery -A proj.celery_app flower

# Custom port
celery -A proj.celery_app flower --port=5555

# Basic auth
celery -A proj.celery_app flower --basic-auth=user:password

Events

# Enable events on worker
celery -A proj.celery_app worker -E

# Terminal event monitor
celery -A proj.celery_app events

Logging

import logging
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@app.task(bind=True)
def rebuild_index(self):
    logger.info("starting rebuild", extra={"task_id": self.request.id})

Custom State

@app.task(bind=True)
def import_rows(self, file_id):
    total = 1000
    for current in range(total):
        ...
        self.update_state(
            state="PROGRESS",
            meta={"current": current, "total": total},
        )

Broker and Backend URLs

Redis

broker_url = "redis://localhost:6379/0"
result_backend = "redis://localhost:6379/1"

# With password
broker_url = "redis://:password@localhost:6379/0"

# TLS
broker_url = "rediss://:password@redis.example.com:6379/0"

RabbitMQ

broker_url = "amqp://guest:guest@localhost:5672//"

# Virtual host
broker_url = "amqp://user:password@rabbitmq.example.com:5672/myvhost"

Database Result Backend

pip install sqlalchemy

result_backend = "db+postgresql://user:password@localhost/celery_results"

Serialization

task_serializer = "json"
result_serializer = "json"
accept_content = ["json"]
timezone = "UTC"
enable_utc = True

Avoid pickle unless every producer, worker, and broker is trusted.

# Do not enable this casually
task_serializer = "pickle"
accept_content = ["pickle"]

Pass simple data:

# Good
send_email.delay(user_id=42)

# Risky: large, stale, or unserializable object
send_email.delay(user_object)

Django Setup

config/celery.py:

import os
from celery import Celery

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

app = Celery("config")
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/1"
CELERY_TASK_SERIALIZER = "json"
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_RESULT_SERIALIZER = "json"
CELERY_TIMEZONE = "UTC"
CELERY_TASK_TRACK_STARTED = True

app/tasks.py:

from celery import shared_task

@shared_task(bind=True)
def send_invoice(self, invoice_id):
    ...

Run:

celery -A config worker --loglevel=INFO
celery -A config beat --loglevel=INFO

Django Transactions

Queue tasks after the database transaction commits.

from django.db import transaction

def create_order(request):
    order = Order.objects.create(...)

    transaction.on_commit(
        lambda: send_order_email.delay(order.id)
    )

FastAPI Setup

worker.py:

from celery import Celery

celery_app = Celery(
    "worker",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

@celery_app.task
def send_email(email: str):
    ...

main.py:

from fastapi import FastAPI
from worker import send_email

app = FastAPI()

@app.post("/send")
def send(email: str):
    result = send_email.delay(email)
    return {"task_id": result.id}

Run:

uvicorn main:app --reload
celery -A worker.celery_app worker --loglevel=INFO

Docker Compose

services:
  redis:
    image: redis:7
    ports:
      - "6379:6379"

  web:
    build: .
    command: uvicorn main:app --host 0.0.0.0 --port 8000
    environment:
      CELERY_BROKER_URL: redis://redis:6379/0
      CELERY_RESULT_BACKEND: redis://redis:6379/1
    depends_on:
      - redis

  worker:
    build: .
    command: celery -A worker.celery_app worker --loglevel=INFO
    environment:
      CELERY_BROKER_URL: redis://redis:6379/0
      CELERY_RESULT_BACKEND: redis://redis:6379/1
    depends_on:
      - redis

  beat:
    build: .
    command: celery -A worker.celery_app beat --loglevel=INFO
    environment:
      CELERY_BROKER_URL: redis://redis:6379/0
    depends_on:
      - redis

Testing

Eager Mode

task_always_eager = True
task_eager_propagates = True

def test_add():
    assert add.delay(2, 3).get() == 5

Unit Test Task Logic

Keep complex business logic in normal functions and call those from tasks.

def build_invoice(invoice_id):
    ...

@app.task
def build_invoice_task(invoice_id):
    return build_invoice(invoice_id)

def test_build_invoice():
    assert build_invoice(invoice.id).total == 100

Mock Task Dispatch

def test_view_queues_task(mocker, client):
    mocked = mocker.patch("app.views.send_email.delay")
    client.post("/signup", data={...})
    mocked.assert_called_once()

Production Settings

broker_connection_retry_on_startup = True
worker_prefetch_multiplier = 1
task_acks_late = True
task_reject_on_worker_lost = True
task_track_started = True
task_time_limit = 600
task_soft_time_limit = 540
result_expires = 3600
worker_max_tasks_per_child = 1000
worker_send_task_events = True
task_send_sent_event = True

Setting	Typical reason
`broker_connection_retry_on_startup`	Keep retrying broker connection during deploy/startup
`result_expires`	Prevent old result buildup
`worker_max_tasks_per_child`	Limit memory growth
`worker_send_task_events`	Enable monitoring events
`task_send_sent_event`	Emit task-sent events from producers

Security

Practice	Why
Use JSON serialization	Avoid arbitrary code execution risk from pickle
Protect broker network access	Anyone with broker write access can enqueue tasks
Use TLS for remote brokers	Prevent credential and payload exposure
Do not pass secrets as args	Task args may appear in logs, events, and results
Set result expiration	Avoid storing sensitive task output indefinitely
Restrict Flower	It can expose task names, args, states, and worker controls

task_serializer = "json"
accept_content = ["json"]
result_expires = 3600

Common Patterns

Fire-and-Forget Task

@app.task(ignore_result=True)
def send_metric(event):
    ...

Dedicated Queue for Slow Tasks

task_routes = {
    "reports.generate": {"queue": "slow"},
}
worker_prefetch_multiplier = 1

celery -A proj.celery_app worker -Q slow --concurrency=2

Rate-Limited Task

@app.task(rate_limit="100/m")
def call_partner_api(payload):
    ...

Singleton-Like Task with Redis Lock

import redis

r = redis.Redis.from_url("redis://localhost:6379/2")

@app.task(bind=True)
def rebuild_cache(self):
    lock = r.lock("lock:rebuild-cache", timeout=600, blocking_timeout=1)
    if not lock.acquire():
        return "already-running"

    try:
        ...
    finally:
        lock.release()

Chunk Large Work

@app.task
def process_batch(ids):
    for item_id in ids:
        process_one(item_id)

for batch in chunks(all_ids, 500):
    process_batch.delay(batch)

Troubleshooting

Symptom	Checks
Task stuck as `PENDING`	Worker not running, wrong broker URL, no result backend, task not imported
`Received unregistered task`	Worker did not import module, wrong `-A`, missing autodiscovery
Task runs twice	Retry/redelivery, worker crash, non-idempotent task, late ack behavior
Worker reserves too much work	Set `worker_prefetch_multiplier=1`
Task never reaches queue	Broker down, producer using wrong app, routing to unconsumed queue
Queue grows forever	Not enough workers, slow task, worker errors, bad routing
Result backend fills up	Set `result_expires`, use `ignore_result=True`
Periodic task runs twice	More than one beat scheduler active
Django task cannot find DB row	Use `transaction.on_commit()`
Memory keeps growing	Add `worker_max_tasks_per_child` or fix leaks

Useful commands:

celery -A proj.celery_app status
celery -A proj.celery_app inspect active
celery -A proj.celery_app inspect reserved
celery -A proj.celery_app inspect scheduled
celery -A proj.celery_app inspect registered
celery -A proj.celery_app report

Redis queue inspection:

redis-cli LLEN celery
redis-cli LRANGE celery 0 5

RabbitMQ queue inspection:

rabbitmqctl list_queues name messages consumers
rabbitmqctl list_exchanges
rabbitmqctl list_bindings

Quick Reference

# Worker
celery -A proj.celery_app worker -l INFO

# Worker for queues
celery -A proj.celery_app worker -Q default,email -l INFO

# Beat
celery -A proj.celery_app beat -l INFO

# Flower
celery -A proj.celery_app flower --port=5555

# Status
celery -A proj.celery_app status

# Inspect
celery -A proj.celery_app inspect active
celery -A proj.celery_app inspect stats

# Revoke
celery -A proj.celery_app control revoke <task-id>

# Send now
task.delay(*args, **kwargs)

# Send with options
task.apply_async(args=[...], kwargs={...}, countdown=60, queue="email")

# Retry
raise self.retry(exc=exc, countdown=30)

# Chain
(task1.s() | task2.s() | task3.s()).delay()

# Group
group(task.s(x) for x in items).delay()

# Chord
chord(group(task.s(x) for x in items), callback.s()).delay()

Celery

On this page