Celery Essentials

Celery is a distributed task queue for Python: it lets you push work (sending an email, resizing an image, running a report, calling a slow third-party API) out of the request/response cycle and onto worker processes that pick it up asynchronously, on a schedule, or in retryable background chains. A web request that would otherwise block for 5 seconds sending an email can instead enqueue a task and return in milliseconds.

Celery needs a message broker to move task messages between the app and workers — Redis and RabbitMQ are the two common choices, Redis being the simpler default for most apps. It optionally uses a result backend (often the same Redis instance, or a database) to store task return values and state if you need to check on a task later.

Setup: app instance and task definitions

Everything starts from a Celery app instance, configured with a broker URL. Tasks are plain Python functions decorated with @app.task — calling .delay() (or the more flexible .apply_async()) sends the call to the broker instead of running it locally.

# celery_app.py
from celery import Celery

app = Celery(
    'myproject',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1',
)

app.conf.update(
    task_serializer='json',
    result_serializer='json',
    accept_content=['json'],
    timezone='UTC',
    enable_utc=True,
    task_acks_late=True,          # ack after the task finishes, not before
    worker_prefetch_multiplier=1,  # don't hoard tasks on one worker
)

# tasks.py
from celery_app import app
import requests

@app.task(bind=True, max_retries=3, default_retry_delay=30)
def send_welcome_email(self, user_id: int):
    try:
        user = get_user(user_id)
        send_email(to=user.email, template='welcome', context={'name': user.name})
    except requests.RequestException as exc:
        # Exponential-ish backoff via countdown; give up after max_retries
        raise self.retry(exc=exc, countdown=2 ** self.request.retries * 10)

@app.task
def generate_report(report_id: int) -> str:
    report = build_report(report_id)
    save_report_pdf(report)
    return report.file_path

Calling a task looks like this from your web app or another task:

from tasks import send_welcome_email, generate_report

# Fire-and-forget
send_welcome_email.delay(user.id)

# Equivalent, but with more control (queue, countdown, expiry)
send_welcome_email.apply_async(args=[user.id], queue='emails', countdown=5)

# If you need the result later
result = generate_report.delay(report_id=42)
result.id          # task UUID, store this if you need to poll later
result.status       # 'PENDING' | 'STARTED' | 'SUCCESS' | 'FAILURE' | 'RETRY'
result.get(timeout=10)  # blocks — avoid calling this from a web request

Workers, queues, and routing

A worker is a separate process that connects to the broker, pulls tasks, and executes them — it's not something the web process runs itself. Start one with the celery CLI, pointed at your app instance:

# Start a worker consuming the default queue, 4 concurrent worker processes
celery -A celery_app worker --loglevel=info --concurrency=4

# Consume specific named queues only
celery -A celery_app worker --loglevel=info --queues=emails,reports

# Celery Beat: the scheduler process for periodic tasks (runs separately from workers)
celery -A celery_app beat --loglevel=info

Routing lets you send different task types to different queues, so you can scale worker pools independently — e.g. a small pool for CPU-light emails and a separate, bigger pool for CPU-heavy report generation:

app.conf.task_routes = {
    'tasks.send_welcome_email': {'queue': 'emails'},
    'tasks.generate_report': {'queue': 'reports'},
}

# Periodic tasks via Celery Beat
from celery.schedules import crontab

app.conf.beat_schedule = {
    'cleanup-expired-sessions': {
        'task': 'tasks.cleanup_expired_sessions',
        'schedule': crontab(hour=3, minute=0),  # daily at 03:00 UTC
    },
    'sync-inventory-every-15-min': {
        'task': 'tasks.sync_inventory',
        'schedule': 300.0,  # seconds — every 5 minutes
    },
}

Chaining, grouping, and error handling

Celery's canvas primitives compose tasks into pipelines. `chain` runs tasks sequentially, feeding each result into the next; `group` runs tasks in parallel; `chord` runs a group and then a callback once every task in the group finishes — the standard pattern for "do N things in parallel, then aggregate."

from celery import chain, group, chord

# Sequential pipeline: resize -> watermark -> upload
pipeline = chain(
    resize_image.s(image_id),
    watermark_image.s(),
    upload_to_cdn.s(),
)
pipeline.apply_async()

# Fan out to N workers in parallel, then aggregate
job = chord(
    (fetch_price.s(symbol) for symbol in ['AAPL', 'MSFT', 'GOOG']),
    aggregate_prices.s(),
)
job.apply_async()

Retries should be explicit and bounded — an unbounded retry on a permanently-broken task (bad input, a deleted foreign key) will loop forever and quietly burn worker capacity. Distinguish transient failures (network timeout — retry) from permanent ones (validation error — fail and alert) rather than retrying everything blindly.

@app.task(bind=True, max_retries=5, autoretry_for=(requests.ConnectionError, requests.Timeout), retry_backoff=True, retry_backoff_max=600, retry_jitter=True)
def sync_inventory(self, warehouse_id: int):
    response = requests.get(f'https://wms.example.com/warehouses/{warehouse_id}', timeout=5)
    response.raise_for_status()
    update_local_inventory(response.json())

Common pitfalls and gotchas

• Task arguments must be JSON-serializable (with the default json serializer) — don't pass ORM model instances or DB sessions into a task; pass IDs and re-fetch inside the task. The object may be stale or unpicklable by the time the worker runs it.

• Tasks should be idempotent where possible — a message can be delivered more than once (worker crash after processing but before ack, `task_acks_late` combined with a crash, broker redelivery). Design tasks so running them twice is safe.

• `.get()` on a result blocks the calling process/thread; calling it from inside a web request re-introduces the synchronous wait Celery exists to avoid. Poll asynchronously or use a callback pattern instead.

• Without a result backend configured, calling `.get()` or checking `.status` raises — decide up front whether you actually need task results persisted, since a result backend adds storage and cleanup overhead (`result_expires`) you don't need for pure fire-and-forget tasks.

• Long-running tasks should report progress or heartbeat rather than running silent for hours — otherwise a hung task and a slow-but-healthy task look identical from the outside.

• `worker_prefetch_multiplier=1` matters for long, uneven tasks: the default prefetches multiple tasks per worker process, which can leave one worker sitting on several slow tasks while other workers are idle.

• Celery Beat and Celery workers are separate processes — running `celery worker` alone will not fire your periodic tasks; you need a `celery beat` process running somewhere too (and only one instance of it, to avoid duplicate schedules).