$devvkit learn --librarie celery-guide

Celery Guide

[python][async][task-queue][distributed]

Python

Install

pip install celery[redis]
# or: uv add celery[redis]

Celery is the most popular task queue for Python. It runs tasks in worker processes (or threads) asynchronously from the main application. Common use cases: sending emails, image processing, data pipelines, cron-style scheduled tasks.

Celery supports multiple brokers: Redis, RabbitMQ, Amazon SQS. Results can be stored in Redis, databases, or ignored. Tasks support retries, rate limiting, time limits, and soft/hard timeouts.

Celery beat runs periodic tasks on a schedule (like cron). Flower is a real-time web UI for monitoring workers and tasks. Celery integrates with Django, Flask, and FastAPI.

Setup

Basic setup— Create Celery app.

from celery import Celery

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

Tasks

Define task— Simple task.

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

@app.task(bind=True, max_retries=3)
def send_welcome_email(self, email: str):
    try:
        mailer.send(email, template="welcome")
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)

Call task— Dispatch asynchronously.

from tasks import add, send_welcome_email

# Fire and forget
add.delay(4, 4)
# With arguments
send_welcome_email.delay("user@example.com")
# Get result
result = add.apply_async(args=[4, 4], queue="high_priority")
result.get(timeout=10)
# Run synchronously (for testing)
add(4, 4)

Task binding— Access task context.

@app.task(bind=True)
def process_file(self, path: str):
    self.logger.info(f"Processing {path}")
    if self.request.retries > 0:
        self.logger.warning(f"Retry #{self.request.retries}")
    # ...
    return {"status": "done"}

Workers

Run worker— Start a worker process.

# Terminal 1: Start worker
celery -A myapp worker --loglevel=info --concurrency=4

# With queues
celery -A myapp worker -Q high_priority,default

# With autoscale
celery -A myapp worker --autoscale=10,3

Configuration

Configuration— Configure Celery.

app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_acks_late=True,
    worker_prefetch_multiplier=1,
    task_soft_time_limit=300,
    task_time_limit=600,
)

Periodic Tasks

Periodic tasks— Celery beat schedule.

from celery.schedules import crontab

app.conf.beat_schedule = {
    "send-daily-summary": {
        "task": "tasks.send_daily_summary",
        "schedule": crontab(hour=8, minute=0),
    },
    "cleanup-every-hour": {
        "task": "tasks.cleanup",
        "schedule": 3600.0,  # seconds
    },
}

# Start beat:
# celery -A myapp beat --loglevel=info

Monitoring

Flower monitoring— Web UI for workers.

# Install
pip install flower

# Run
celery -A myapp flower --port=5555

# Open http://localhost:5555