Celery

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()