Celery and RQ — An Introduction

Capsule overview

FastAPI's BackgroundTasks is perfect for lightweight operations: sending an email, writing a log, invalidating a cache. But it has fundamental limitations: the tasks run in the same process as your API, they're lost if the server restarts, there's no automatic retry, you can't schedule tasks to run in the future, and you can't distribute the work across multiple servers.

When you need more than that — heavy processing, automatic retries, scheduling, monitoring, persistence — you need an external task queue. The two most popular in Python are Celery (the industry standard, powerful but complex) and RQ (Redis Queue, simple and lightweight). This capsule is a conceptual introduction with basic examples — not a complete setup guide. The goal is for you to understand when you need to scale beyond BackgroundTasks and which tool to pick.

The limitations of BackgroundTasks

Before you look at the alternatives, understand exactly where BackgroundTasks falls short:

When is BackgroundTasks fine?

• ✅ Sending emails/notifications (if losing one isn't critical)
• ✅ Logging and auditing
• ✅ Invalidating caches
• ✅ Operations that take < 30 seconds
• ✅ Tasks where losing one isn't catastrophic (it can be re-run)

When do you need a task queue?

• ❌ Image/video processing (heavy, takes minutes)
• ❌ Bulk email sending (thousands at a time)
• ❌ Generating large reports (lots of CPU/memory)
• ❌ Tasks that MUST complete (payments, legal confirmations)
• ❌ Scheduled tasks (cron jobs: "clean up data every night")
• ❌ Spreading the load across multiple servers

The architecture of a task queue

A task queue has three components:

┌─────────┐     ┌─────────┐     ┌─────────┐
│   API   │ ──▶ │ Broker  │ ──▶ │ Worker  │
│ (FastAPI)│     │ (Redis) │     │ (Celery)│
└─────────┘     └─────────┘     └─────────┘
     │               │               │
  Sends task    Stores it in    Reads and runs
  to the broker  a queue        the task

1. Producer (your FastAPI app): creates the task and sends it to the broker
2. Broker (Redis, RabbitMQ): stores the tasks in a queue
3. Worker (Celery/RQ): reads tasks from the queue and runs them

The key difference from BackgroundTasks: the worker is a separate process. If your API restarts, the tasks in the queue are still there. If the worker fails, the task gets re-queued. If you need more capacity, you add more workers.

Redis as the broker

Both Celery and RQ use Redis as the broker (the message queue). Redis is an in-memory database that acts as the middleman between your API and the workers.

What is Redis?

Redis is an in-memory data store, extremely fast. In the context of task queues:

• As a broker: it stores tasks in FIFO queues (First In, First Out)
• As a result backend: it stores the results of completed tasks
• As a cache: it stores temporary, frequently accessed data

Installing Redis (conceptually)

# macOS
brew install redis
brew services start redis

# Docker (recommended for development)
docker run -d --name redis -p 6379:6379 redis:alpine

# Check that it's running
redis-cli ping
# → PONG

Redis runs on localhost:6379 by default. Both Celery and RQ connect to that address.

Celery: the industry standard

Celery is the most popular task queue in Python. It's powerful, mature, and supports advanced features: automatic retries, scheduling with Celery Beat, task routing, monitoring with Flower, and more.

Celery's architecture

FastAPI App                     Redis Broker              Celery Worker(s)
┌──────────────┐               ┌──────────┐              ┌──────────────┐
│ POST /tasks  │ ── task ──▶  │  Queue   │ ── task ──▶ │ @celery.task │
│              │               │          │              │ def process():│
│ return 202   │               │  tasks   │              │   ...        │
└──────────────┘               └──────────┘              └──────────────┘
                                    │                          │
                              Persists tasks              Runs tasks
                              in the queue                independently

Installation

pip install celery redis

A basic Celery example

worker.py — defines the tasks:

from celery import Celery

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

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


@celery_app.task(bind=True, max_retries=3, default_retry_delay=60)
def send_email_task(self, to: str, subject: str, body: str):
    """A Celery task for sending an email."""
    try:
        print(f"Sending email to {to}: {subject}")
        # The real sending logic would go here
        return {"status": "sent", "to": to}

    except Exception as exc:
        print(f"Failed to send email, retrying... ({self.request.retries}/{self.max_retries})")
        raise self.retry(exc=exc)


@celery_app.task
def generate_report_task(report_type: str, user_id: int):
    """A heavy report generation task."""
    import time
    time.sleep(30)
    return {"report_type": report_type, "status": "generated", "user_id": user_id}

app/main.py — the FastAPI integration:

from fastapi import FastAPI
from worker import send_email_task, generate_report_task

app = FastAPI()


@app.post("/tasks", status_code=201)
def create_task(title: str, assignee_email: str):
    task = {"id": 1, "title": title}

    result = send_email_task.delay(
        to=assignee_email,
        subject=f"New task: {title}",
        body=f"You were assigned the task '{title}'",
    )

    return {
        "task": task,
        "notification": {
            "celery_task_id": result.id,
            "status": "queued",
        },
    }


@app.post("/reports/generate", status_code=202)
def generate_report(report_type: str, user_id: int):
    result = generate_report_task.delay(report_type, user_id)

    return {
        "message": "Report generation started",
        "celery_task_id": result.id,
        "status_url": f"/reports/status/{result.id}",
    }


@app.get("/reports/status/{task_id}")
def get_report_status(task_id: str):
    from celery.result import AsyncResult

    result = AsyncResult(task_id)

    return {
        "task_id": task_id,
        "status": result.status,
        "result": result.result if result.ready() else None,
    }

Running Celery

# Terminal 1: Redis
redis-server

# Terminal 2: Celery Worker
celery -A worker worker --loglevel=info

# Terminal 3: FastAPI
uvicorn app.main:app --reload

The anatomy of a Celery task

@celery_app.task(
    bind=True,              # self = the task instance
    max_retries=3,          # retry up to 3 times
    default_retry_delay=60, # wait 60s between retries
    time_limit=300,         # a 5-minute timeout
    soft_time_limit=240,    # a warning at 4 minutes
)
def my_task(self, arg1, arg2):
    try:
        # ... work ...
        return {"status": "done"}
    except Exception as exc:
        raise self.retry(exc=exc)

.delay() vs .apply_async()

# .delay() — the simple form
send_email_task.delay("ana@test.com", "Subject", "Body")

# .apply_async() — full control
send_email_task.apply_async(
    args=["ana@test.com", "Subject", "Body"],
    countdown=60,          # run it in 60 seconds
    expires=3600,          # expire in 1 hour if it hasn't run
    retry=True,            # automatic retry
    retry_policy={
        "max_retries": 5,
        "interval_start": 10,  # first retry after 10s
        "interval_step": 20,   # each retry +20s
    },
)

RQ (Redis Queue): the lightweight alternative

RQ is simpler than Celery. Fewer features, but easier to learn and configure. Ideal for small-to-medium projects.

Installation

pip install rq

A basic RQ example

tasks.py — defines the functions (plain functions, no decorators):

import time


def send_email(to: str, subject: str):
    """A plain function — RQ runs it as a task."""
    print(f"Sending email to {to}: {subject}")
    time.sleep(2)
    return {"status": "sent", "to": to}


def generate_report(report_type: str):
    """A heavy report generation task."""
    print(f"Generating {report_type} report...")
    time.sleep(30)
    return {"report_type": report_type, "status": "generated"}

app/main.py — the FastAPI integration:

from fastapi import FastAPI
from redis import Redis
from rq import Queue
from tasks import send_email, generate_report

app = FastAPI()

redis_conn = Redis(host="localhost", port=6379)
queue = Queue(connection=redis_conn)


@app.post("/tasks", status_code=201)
def create_task(title: str, email: str):
    task = {"id": 1, "title": title}

    job = queue.enqueue(send_email, email, f"New task: {title}")

    return {
        "task": task,
        "notification": {"job_id": job.id, "status": job.get_status()},
    }


@app.get("/jobs/{job_id}")
def get_job_status(job_id: str):
    from rq.job import Job

    try:
        job = Job.fetch(job_id, connection=redis_conn)
        return {
            "job_id": job_id,
            "status": job.get_status(),
            "result": job.result,
        }
    except Exception:
        return {"job_id": job_id, "status": "not_found"}

Running RQ

# Terminal 1: Redis
redis-server

# Terminal 2: RQ Worker
rq worker

# Terminal 3: FastAPI
uvicorn app.main:app --reload

The key difference from Celery

In RQ, tasks are plain Python functions. They don't need decorators or a celery_app object. Any importable function can be an RQ task:

# Celery: needs a decorator
@celery_app.task
def my_task(x):
    return x * 2

# RQ: a plain function
def my_task(x):
    return x * 2

# Enqueue in RQ
queue.enqueue(my_task, 42)

Comparison: BackgroundTasks vs Celery vs RQ

Which one should you pick?

Does the task take < 30 seconds?
├── Yes → Is it critical (it can't be lost)?
│         ├── No → BackgroundTasks ✓
│         └── Yes → RQ or Celery
└── No → Do you need scheduling, complex retries, multiple workers?
         ├── No → RQ ✓
         └── Yes → Celery ✓

In short:

• BackgroundTasks for the simple and fast stuff (90% of cases in small-to-medium APIs)
• RQ when you need persistence and retries without Celery's complexity
• Celery when you need the full toolkit: scheduling, routing, monitoring, distribution

Celery Beat: scheduled tasks

Celery Beat lets you schedule recurring tasks (like cron jobs):

from celery import Celery
from celery.schedules import crontab

celery_app = Celery("tasks", broker="redis://localhost:6379/0")

celery_app.conf.beat_schedule = {
    "cleanup-every-night": {
        "task": "tasks.cleanup_old_tasks",
        "schedule": crontab(hour=2, minute=0),
    },
    "send-daily-report": {
        "task": "tasks.generate_daily_report",
        "schedule": crontab(hour=8, minute=0, day_of_week="mon-fri"),
    },
    "health-check-every-5-min": {
        "task": "tasks.health_check",
        "schedule": 300.0,
    },
}


@celery_app.task
def cleanup_old_tasks():
    """Runs every night at 2 AM."""
    print("Cleaning up old tasks...")


@celery_app.task
def generate_daily_report():
    """Runs Monday through Friday at 8 AM."""
    print("Generating daily report...")


@celery_app.task
def health_check():
    """Runs every 5 minutes."""
    print("Health check OK")

celery -A worker beat --loglevel=info

BackgroundTasks can't do this. If you need scheduled tasks, it's Celery Beat or rq-scheduler.

A real integration pattern: FastAPI + Celery

In a real project, the structure looks like this:

project/
├── app/
│   ├── main.py           ← the FastAPI app
│   ├── routers/
│   │   └── tasks.py      ← endpoints
│   └── dependencies.py
├── worker/
│   ├── celery_app.py     ← Celery configuration
│   ├── tasks/
│   │   ├── email.py      ← email tasks
│   │   └── reports.py    ← report tasks
│   └── config.py         ← broker configuration
├── docker-compose.yml    ← Redis + worker + API
└── requirements.txt

The FastAPI endpoints call .delay() to enqueue tasks. The Celery workers run them independently. Redis acts as the middleman. Docker Compose orchestrates the three services.

This architecture gets covered in detail in later guides in the path. Here the goal is for you to understand the concept and when you need to scale.

Connection with the project

This module's project (Capsule 05) uses BackgroundTasks — not Celery or RQ. The reason is that the Task Manager's operations (simulated notifications, cleanup) are light and perfectly acceptable for BackgroundTasks. But now you know that if the Task Manager grew into production with real emails, PDF generation, and scheduled tasks, you'd need to migrate to Celery or RQ.

Troubleshooting

Problem 1: "Connection refused" when connecting to Redis

Cause: Redis isn't running.

Fix:

# Check whether Redis is running
redis-cli ping
# If it doesn't answer PONG, start Redis:

# macOS
brew services start redis

# Docker
docker run -d --name redis -p 6379:6379 redis:alpine

Problem 2: The Celery worker can't find the tasks

Cause: The tasks module isn't importable from wherever the worker is running.

Fix: Run the worker from the project's root directory:

# ❌ From a subdirectory — it can't find the tasks
cd app && celery -A worker worker

# ✅ From the project root
celery -A worker worker --loglevel=info

Problem 3: The tasks get queued but never run

Cause: There are no workers running, or the workers are connected to a different broker/queue.

Fix: Check that the worker and the app use the same broker URL and that the worker is alive:

# Check for active workers
celery -A worker inspect active

# If there are none, start one
celery -A worker worker --loglevel=info

Problem 4: An RQ job shows "failed" with no error detail

Cause: The enqueued function raised an exception that RQ swallowed.

Fix: Inspect the failed jobs:

from rq import Queue
from redis import Redis

q = Queue(connection=Redis())
failed = q.failed_job_registry

for job_id in failed.get_job_ids():
    job = Job.fetch(job_id, connection=Redis())
    print(f"Job {job_id}: {job.exc_info}")

Problem 5: I don't know whether to use BackgroundTasks or Celery for my case

Cause: The boundary isn't clear.

Fix: Use this rule:

Does the task take < 30 seconds AND is it acceptable to lose it if the server restarts?
→ Yes: BackgroundTasks
→ No: Celery/RQ

If you start with BackgroundTasks and later need to scale, the migration is relatively simple: you extract the function into a separate module, decorate it with @celery_app.task, and swap background_tasks.add_task(func, args) for func.delay(args).

Exercises

Exercise 1: A decision diagram (Easy)

For each scenario, say which tool you'd use (BackgroundTasks, Celery, RQ) and why:

1. Logging every request to a file after responding
2. Sending 10,000 newsletter emails
3. Generating a thumbnail of an uploaded image (takes 5 seconds)
4. Running a database backup every night at 3 AM
5. Invalidating the Redis cache after updating a resource

Exercise 2: A mental model of Celery (Easy)

Draw or describe the data flow when a FastAPI app creates a Celery task. Include: (1) what happens in the API, (2) what happens in Redis, (3) what happens in the worker, (4) how the API checks the result.

1. API (FastAPI):
   - The endpoint receives the request
   - It calls task.delay(args) → serializes the args as JSON
   - The Celery client sends the message to Redis
   - The endpoint returns 202 Accepted with the task_id

2. Redis (Broker):
   - Receives the message in the "celery" queue
   - The message contains: task name, args, kwargs, task_id
   - It stores it in a FIFO queue (a Redis list)
   - It persists it in memory (and optionally on disk)

3. Worker (Celery):
   - It's listening to the queue in Redis (polling)
   - It reads the next message from the queue
   - It deserializes the args
   - It runs the function decorated with @celery_app.task
   - On success: it saves the result in Redis (the backend)
   - On failure: it re-queues according to the retry policy

4. The API checks the result:
   - The GET /status/{task_id} endpoint creates AsyncResult(task_id)
   - AsyncResult reads the result from Redis (the backend)
   - It returns: status (PENDING/STARTED/SUCCESS/FAILURE) + result

Exercise 3: Migrating from BackgroundTasks to Celery (Medium)

You have this endpoint with BackgroundTasks. Write what it would look like migrated to Celery (just the task and the endpoint, not the full setup):

@app.post("/orders", status_code=201)
def create_order(product: str, email: str, background_tasks: BackgroundTasks):
    order = {"id": 1, "product": product}
    background_tasks.add_task(send_confirmation, email, order)
    return order

from worker.celery_app import celery_app


@celery_app.task(bind=True, max_retries=3, default_retry_delay=30)
def send_confirmation(self, email: str, order: dict):
    try:
        print(f"Sending confirmation to {email} for order #{order['id']}")
        return {"status": "sent", "email": email}
    except Exception as exc:
        raise self.retry(exc=exc)

from fastapi import APIRouter
from worker.tasks.orders import send_confirmation

router = APIRouter()


@router.post("/orders", status_code=201)
def create_order(product: str, email: str):
    order = {"id": 1, "product": product}

    result = send_confirmation.delay(email, order)

    return {
        "order": order,
        "notification_task_id": result.id,
    }

Exercise 4: RQ vs Celery for your project (Medium)

You're building an e-commerce platform with these background processing requirements:

• Send a confirmation email when an order is created
• Generate a PDF invoice (takes ~10 seconds)
• Update the inventory after a purchase
• Send a weekly newsletter to 50,000 users
• Clean up expired sessions every hour

Would you use RQ or Celery? Justify it for each task.

Exercise 5: Designing Celery status tracking (Hard)

Design a GET /tasks/{celery_task_id}/status endpoint that uses Celery's AsyncResult to return a task's state. The response should include: task_id, status (PENDING/STARTED/SUCCESS/FAILURE/RETRY), result (if it completed), error (if it failed), progress (if the task reports progress). Write the complete endpoint.

from fastapi import FastAPI
from celery.result import AsyncResult
from worker.celery_app import celery_app

app = FastAPI()


@app.get("/tasks/{task_id}/status")
def get_task_status(task_id: str):
    result = AsyncResult(task_id, app=celery_app)

    response = {
        "task_id": task_id,
        "status": result.status,
        "result": None,
        "error": None,
        "progress": None,
    }

    if result.ready():
        if result.successful():
            response["result"] = result.result
        else:
            response["error"] = str(result.result)

    elif result.status == "PROGRESS":
        response["progress"] = result.info

    return response

@celery_app.task(bind=True)
def long_task(self, total_items):
    for i in range(total_items):
        self.update_state(
            state="PROGRESS",
            meta={"current": i + 1, "total": total_items},
        )
        time.sleep(1)
    return {"status": "completed", "total_processed": total_items}

Summary

• BackgroundTasks is enough for light tasks (< 30s) where losing one isn't critical
• Celery is the standard for production: retries, scheduling, distribution, monitoring
• RQ is the lightweight alternative: same concept, fewer features, easier to learn
• Redis acts as the broker (the message queue) between the API and the workers
• The architecture is: Producer (API) → Broker (Redis) → Worker (Celery/RQ)
• Celery Beat enables scheduled tasks (cron jobs) — BackgroundTasks can't
• .delay() in Celery is the equivalent of add_task() in BackgroundTasks
• RQ uses plain functions — they need no special decorators
• The migration from BackgroundTasks to Celery is mechanical: decorate the function, swap add_task for .delay()
• The decision: < 30s + not critical → BackgroundTasks. Retries + scheduling → Celery. Simple + persistence → RQ

Additional resources

1. FastAPI - Background Tasks — When BackgroundTasks is enough
2. Celery - First Steps — The official Celery tutorial
3. Celery - User Guide — The complete Celery guide
4. RQ Documentation — RQ's official documentation
5. Redis Documentation — Redis's official documentation
6. Flower - Celery Monitor — A monitoring dashboard for Celery
7. Full Stack FastAPI + Celery — A complete integration tutorial

Next capsule: Project: Background Processing — You'll integrate background tasks into your Task Manager API with notifications, cleanup, and status tracking.