Event-Driven Order Processing Pipeline

Build a multi-service order pipeline with Django, Celery, RabbitMQ, and Redis.

Introduction

30 Second Summary

When you order something online, it feels instant. Behind the scenes, that single click triggers a chain of steps across multiple systems, each handling a piece of the process independently.

In this project, you will build an event-driven order processing pipeline running entirely on your local machine. Orders arrive via async API endpoints, flow through a message queue, and get processed through a multi-step pipeline with real-time monitoring.

What You'll Build

An event-driven order processing system with Django ASGI serving async REST endpoints, Celery task chains processing orders through RabbitMQ, Redis caching for fast status lookups, and PostgreSQL for durable storage.

By the end of this project, you'll have:

  • A Docker Compose stack running 6 services that handle async web serving, task processing, message brokering, caching, and scheduled maintenance.
  • Async Django REST endpoints for order creation and status polling with Redis-cached responses for fast lookups.
  • A Celery task chain implementing the full order processing pipeline with automatic retries, exponential backoff, and failure handling.
  • Secret Mission: Add dead letter queue handling and idempotency keys to prevent duplicate order processing.

Are there any prerequisites?

This is a hard project. You should be familiar with Django, Python, and Docker basics before starting.

You will also need Docker Desktop installed and running on your machine.

Set Up Docker Compose Infrastructure

An event-driven pipeline needs multiple services working together. Your system will use PostgreSQL for durable storage, RabbitMQ as the message broker, and Redis for caching and task results.

Instead of installing each service manually, you'll define the entire stack in a single Docker Compose file. This lets you spin up all six services with one command and ensures everyone gets the same environment.

In this step, get ready to:
  • Create the Docker Compose file defining all 6 services.
  • Create the Dockerfile for the Django application container.
  • Start the infrastructure services and verify they are healthy.
Create the Docker Compose file

Docker Compose lets you define multiple containers as services in a single YAML file. Each service gets its own image, ports, health checks, and dependency rules.

  • Create a file called docker-compose.yml in the root of your project folder.
  • Paste the following configuration into it:
services:
  postgres:
    image: postgres:17
    restart: always
    environment:
      POSTGRES_DB: orders_db
      POSTGRES_USER: orders_user
      POSTGRES_PASSWORD: orders_pass
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U orders_user -d orders_db"]
      interval: 10s
      timeout: 5s
      retries: 5

  rabbitmq:
    image: rabbitmq:4.3-management
    restart: always
    ports:
      - "5672:5672"
      - "15672:15672"
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "check_running"]
      interval: 10s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    restart: always
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5

  web:
    build: .
    command: uvicorn core.asgi:application --host 0.0.0.0 --port 8000 --reload
    volumes:
      - .:/app
    ports:
      - "8000:8000"
    env_file:
      - .env
    depends_on:
      postgres:
        condition: service_healthy
      rabbitmq:
        condition: service_healthy
      redis:
        condition: service_healthy

  celery_worker:
    build: .
    command: celery -A core worker --loglevel=info --concurrency=4
    volumes:
      - .:/app
    env_file:
      - .env
    depends_on:
      postgres:
        condition: service_healthy
      rabbitmq:
        condition: service_healthy
      redis:
        condition: service_healthy

  celery_beat:
    build: .
    command: celery -A core beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler
    volumes:
      - .:/app
    env_file:
      - .env
    depends_on:
      postgres:
        condition: service_healthy
      rabbitmq:
        condition: service_healthy
      redis:
        condition: service_healthy

  flower:
    build: .
    command: celery -A core flower --port=5555
    volumes:
      - .:/app
    ports:
      - "5555:5555"
    env_file:
      - .env
    depends_on:
      - celery_worker

volumes:
  postgres_data:

What does this file define?

This Compose file orchestrates six services that form your order processing system.

  • postgres stores orders durably. Its data persists in a named volume (postgres_data) so nothing is lost when the container restarts.
  • rabbitmq is the message broker. The 4.3-management tag includes a web UI for monitoring queues on port 15672.
  • redis serves as both the Celery result backend and the order status cache.
  • web, celery_worker, celery_beat, and flower are your application services. They all build: . from the same Dockerfile and wait for the infrastructure services to be healthy before starting.

Why RabbitMQ instead of Redis as the broker?

RabbitMQ persists messages to disk by default. For an order processing system, losing a message means losing a customer's order. Redis is faster for small payloads, but it can lose data under memory pressure.

Redis still handles the result backend and cache layer where speed matters more than durability.

  • Save docker-compose.yml.

You won't see a change yet. The Compose file references a Dockerfile and .env file that don't exist yet. You'll create those next.

✔️ Awesome, I've got everything!

Great. Make sure the file is saved before moving on.

ⓧ I'd like to double check the full code

services:
  postgres:
    image: postgres:17
    restart: always
    environment:
      POSTGRES_DB: orders_db
      POSTGRES_USER: orders_user
      POSTGRES_PASSWORD: orders_pass
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U orders_user -d orders_db"]
      interval: 10s
      timeout: 5s
      retries: 5

  rabbitmq:
    image: rabbitmq:4.3-management
    restart: always
    ports:
      - "5672:5672"
      - "15672:15672"
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "check_running"]
      interval: 10s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    restart: always
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5

  web:
    build: .
    command: uvicorn core.asgi:application --host 0.0.0.0 --port 8000 --reload
    volumes:
      - .:/app
    ports:
      - "8000:8000"
    env_file:
      - .env
    depends_on:
      postgres:
        condition: service_healthy
      rabbitmq:
        condition: service_healthy
      redis:
        condition: service_healthy

  celery_worker:
    build: .
    command: celery -A core worker --loglevel=info --concurrency=4
    volumes:
      - .:/app
    env_file:
      - .env
    depends_on:
      postgres:
        condition: service_healthy
      rabbitmq:
        condition: service_healthy
      redis:
        condition: service_healthy

  celery_beat:
    build: .
    command: celery -A core beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler
    volumes:
      - .:/app
    env_file:
      - .env
    depends_on:
      postgres:
        condition: service_healthy
      rabbitmq:
        condition: service_healthy
      redis:
        condition: service_healthy

  flower:
    build: .
    command: celery -A core flower --port=5555
    volumes:
      - .:/app
    ports:
      - "5555:5555"
    env_file:
      - .env
    depends_on:
      - celery_worker

volumes:
  postgres_data:
Create the Dockerfile for the Django application

The four application services (web, celery_worker, celery_beat, flower) all share the same Docker image. The Dockerfile defines how to build that image from a Python base.

  • Create a file called Dockerfile (no extension) in your project folder.
  • Paste the following into it:
FROM python:3.12-slim

ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

What does this Dockerfile do?

  • FROM python:3.12-slim starts from a lightweight Python 3.12 image.
  • PYTHONDONTWRITEBYTECODE=1 prevents Python from writing .pyc files, keeping the image clean.
  • PYTHONUNBUFFERED=1 ensures logs appear immediately in docker compose logs instead of being buffered.
  • Requirements are copied and installed before the rest of the code. This means Docker can cache the dependency layer and only reinstall when requirements.txt changes.
  • Save Dockerfile.
Create the dependency and environment files, then start infrastructure

Before Docker can build the application image, it needs to know which Python packages to install. You also need an environment file so the services know how to connect to each other.

  • Create a file called requirements.txt in your project folder and paste the following pinned dependencies:
Django==6.0.6
celery==5.6.3
uvicorn==0.49.0
django-celery-beat==2.9.0
django-redis==7.0.0
psycopg[binary]==3.3.4
djangorestframework==3.16.0
flower==2.0.1
redis==5.3.0

What are these packages?

  • Django is the web framework. uvicorn serves it as an ASGI app, enabling async views.
  • celery handles background task processing. django-celery-beat adds periodic task scheduling. flower provides a real-time monitoring dashboard.
  • psycopg[binary] is the modern PostgreSQL adapter for Python. django-redis integrates Redis with Django's cache framework.
  • djangorestframework gives you serializers and a browsable API for your REST endpoints.
  • Save requirements.txt.
  • Create a file called .env in your project folder and paste the following connection strings:
DJANGO_SETTINGS_MODULE=core.settings
DATABASE_URL=postgres://orders_user:orders_pass@postgres:5432/orders_db
CELERY_BROKER_URL=amqp://guest:guest@rabbitmq:5672//
CELERY_RESULT_BACKEND=redis://redis:6379/0
REDIS_URL=redis://redis:6379/1

What are these environment variables?

These connection strings tell each service where to find the others. The hostnames (postgres, rabbitmq, redis) match the service names in docker-compose.yml. Docker Compose automatically creates a network where services can reach each other by name.

  • Save .env.

Now you have all four files in place. Before building the Django app image, start just the infrastructure services to confirm they come up healthy.

  • Start PostgreSQL, RabbitMQ, and Redis by running this command in your project folder:
docker compose up -d postgres rabbitmq redis

What does this command do?

The -d flag runs containers in the background (detached mode). By naming only postgres rabbitmq redis, you start just the infrastructure services without trying to build the Django image yet.

  • Wait about 15-20 seconds for the health checks to pass, then verify the services are healthy by running:
docker compose ps

You should see all three services listed with a status of healthy in the output.

  • Open your browser and go to http://localhost:15672.
  • Log in with username guest and password guest.

You should see the RabbitMQ Management dashboard showing an overview of your broker with zero queues and connections so far.

  • Verify Redis is responding by running:
docker compose exec redis redis-cli ping

You should see PONG in the terminal output.

Services not showing as healthy?

  • Make sure Docker Desktop is running. You can check by looking for the Docker icon in your system tray or menu bar.
  • If a service is in a restart loop, check its logs by running docker compose logs postgres (replace with the service name that's failing).
  • If port 5432, 5672, or 6379 is already in use by another process, stop that process or change the host port in docker-compose.yml.

Help me debug why my Docker Compose services aren't starting.

Your infrastructure backbone is live. PostgreSQL is ready to store orders, RabbitMQ is waiting for messages, and Redis is standing by to cache. Next up, you'll build the Django application that ties them all together.

Configure Django ASGI Project and Celery

Your infrastructure services are healthy and waiting for an application to connect to them. The next piece of the puzzle is the Django application itself, configured to run asynchronously with Uvicorn and dispatch tasks to Celery through RabbitMQ.

In this step, you will create the Django project structure, configure Celery to use RabbitMQ as its broker and Redis as its result backend, define an Order model with a status state machine, and set up Django REST Framework for the API layer.

In this step, get ready to:
  • Create the Django ASGI project with Celery app initialization and full settings.
  • Define an Order model with a status state machine (pending, processing, completed, failed).
  • Configure Django REST Framework serializers and URL routing.
Create the Django project with ASGI and Celery configuration

The Docker Compose file references core.asgi:application as the entry point. You need to create this Django project structure manually so every service in the stack can find the application.

  • Create the project directories by running these commands in your terminal (inside the order-pipeline folder):
mkdir core
mkdir orders
  • Confirm the directories exist by running:
ls

You should see core and orders alongside your existing files (docker-compose.yml, Dockerfile, etc.).

  • Create manage.py in the root of your project folder with this content:
#!/usr/bin/env python
import os
import sys


def main():
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "core.settings")
    try:
        from django.core.management import execute_from_command_line
    except ImportError as exc:
        raise ImportError(
            "Couldn't import Django. Are you sure it's installed and "
            "available on your PYTHONPATH environment variable? Did you "
            "forget to activate a virtual environment?"
        ) from exc
    execute_from_command_line(sys.argv)


if __name__ == "__main__":
    main()

What does manage.py do?

This is Django's standard command-line utility. It sets the DJANGO_SETTINGS_MODULE environment variable so Django knows where your project configuration lives, then delegates to Django's management commands (migrate, createsuperuser, etc.).

  • Create core/celery.py to initialize the Celery application:
import os

from celery import Celery

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

app = Celery("core")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()
  • Create core/__init__.py to ensure Django loads the Celery app on startup:
from .celery import app as celery_app

__all__ = ("celery_app",)

What does this Celery setup do?

The celery.py file creates a Celery application instance and tells it to read configuration from Django's settings (any setting prefixed with CELERY_). The autodiscover_tasks() call scans all installed apps for a tasks.py file so Celery finds your task functions automatically.

Importing the Celery app in __init__.py guarantees the app is loaded whenever Django starts, which is required for the @shared_task decorator to work correctly.

  • Create core/asgi.py as the ASGI entry point for Uvicorn:
import os

from django.core.asgi import get_asgi_application

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

application = get_asgi_application()

Why ASGI instead of WSGI?

ASGI (Asynchronous Server Gateway Interface) allows Django to handle requests asynchronously. Uvicorn serves this ASGI application, enabling async views that won't block the event loop while waiting on database queries or cache reads.

  • Create core/settings.py with the full project configuration:
import os
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent

SECRET_KEY = "dev-secret-key-change-in-production"
DEBUG = True
ALLOWED_HOSTS = ["*"]

INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    "rest_framework",
    "django_celery_beat",
    "orders",
]

MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "django.contrib.sessions.middleware.SessionMiddleware",
    "django.middleware.common.CommonMiddleware",
    "django.middleware.csrf.CsrfViewMiddleware",
    "django.contrib.auth.middleware.AuthenticationMiddleware",
    "django.contrib.messages.middleware.MessageMiddleware",
    "django.middleware.clickjacking.XFrameOptionsMiddleware",
]

ROOT_URLCONF = "core.urls"

TEMPLATES = [
    {
        "BACKEND": "django.template.backends.django.DjangoTemplates",
        "DIRS": [],
        "APP_DIRS": True,
        "OPTIONS": {
            "context_processors": [
                "django.template.context_processors.request",
                "django.contrib.auth.context_processors.auth",
                "django.contrib.messages.context_processors.messages",
            ],
        },
    },
]

ASGI_APPLICATION = "core.asgi.application"

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": "orders_db",
        "USER": "orders_user",
        "PASSWORD": "orders_pass",
        "HOST": "postgres",
        "PORT": "5432",
    }
}

CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": os.environ.get("REDIS_URL", "redis://redis:6379/1"),
        "OPTIONS": {
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
        },
    }
}

CELERY_BROKER_URL = os.environ.get("CELERY_BROKER_URL", "amqp://guest:guest@rabbitmq:5672//")
CELERY_RESULT_BACKEND = os.environ.get("CELERY_RESULT_BACKEND", "redis://redis:6379/0")
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERY_TIMEZONE = "UTC"
CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers:DatabaseScheduler"

REST_FRAMEWORK = {
    "DEFAULT_RENDERER_CLASSES": [
        "rest_framework.renderers.JSONRenderer",
        "rest_framework.renderers.BrowsableAPIRenderer",
    ],
}

STATIC_URL = "static/"
DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"

What's happening in these settings?

  • DATABASES connects to the PostgreSQL container using the credentials from docker-compose.yml. The host is postgres (the Docker service name).
  • CACHES configures django-redis to use Redis database 1 for caching (database 0 is reserved for Celery results).
  • CELERY_BROKER_URL points Celery at RabbitMQ for message transport. CELERY_RESULT_BACKEND stores task results in Redis for fast lookups.
  • CELERY_BEAT_SCHEDULER uses the database-backed scheduler from django-celery-beat so scheduled tasks persist across restarts.
  • Save all four files in the core/ directory and manage.py in the project root.

Files not in the right place?

Make sure core/ is a directory (not a file) and that __init__.py, celery.py, asgi.py, and settings.py are directly inside it. The manage.py file lives at the same level as docker-compose.yml, not inside core/.

help me figure out why Django can't find my settings module

Create the orders app with the Order model

The orders app holds the Order model that tracks each order through its lifecycle. The model uses a status field as a state machine with four states: pending, processing, completed, and failed.

  • Create an empty orders/__init__.py file to make orders a Python package. You can leave this file completely empty.
  • Create orders/models.py with the Order model:
from django.db import models


class Order(models.Model):
    STATUS_CHOICES = [
        ("pending", "Pending"),
        ("processing", "Processing"),
        ("completed", "Completed"),
        ("failed", "Failed"),
    ]

    customer_email = models.EmailField()
    items = models.JSONField()
    total_amount = models.DecimalField(max_digits=10, decimal_places=2)
    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default="pending")
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    failure_reason = models.TextField(blank=True, default="")

    class Meta:
        ordering = ["-created_at"]

    def __str__(self):
        return f"Order {self.id} - {self.status}"

What does this model do?

  • STATUS_CHOICES defines the four states an order can be in. The status field defaults to pending when a new order is created.
  • items uses a JSONField to store a flexible list of products without needing a separate line-items table.
  • failure_reason captures why an order failed (payment timeout, validation error, etc.) so you can debug issues without digging through logs.
  • updated_at with auto_now=True timestamps every status change, which the cleanup task uses to detect stale orders.
  • Save orders/__init__.py and orders/models.py.

Not sure if the model is correct?

Double-check that STATUS_CHOICES is a class-level attribute (inside the class but not inside a method). Each choice is a tuple of (database_value, display_label).

help me check my Order model definition is correct

Configure serializers and URL routing

Django REST Framework needs serializers to convert Order instances to JSON and validate incoming data. You also need URL routing to connect incoming HTTP requests to the correct views.

  • Create orders/serializers.py with two serializers for creating and reading orders:
from rest_framework import serializers

from .models import Order


class OrderCreateSerializer(serializers.ModelSerializer):
    class Meta:
        model = Order
        fields = ["customer_email", "items", "total_amount"]


class OrderStatusSerializer(serializers.ModelSerializer):
    class Meta:
        model = Order
        fields = ["id", "status", "created_at", "updated_at", "failure_reason"]

Why two serializers?

OrderCreateSerializer only accepts the fields a customer provides (email, items, amount). It prevents clients from setting status or timestamps directly.

OrderStatusSerializer exposes the fields needed for status polling. Keeping them separate follows the principle of least privilege for your API.

  • Create orders/urls.py to define the API endpoints:
from django.urls import path

from .views import OrderCreateView, OrderStatusView

urlpatterns = [
    path("orders/", OrderCreateView.as_view(), name="order-create"),
    path("orders/<int:order_id>/status/", OrderStatusView.as_view(), name="order-status"),
]
  • Create orders/views.py with placeholder view classes (you will implement these fully in the next step):
from django.views import View


# Placeholder views - full implementation in the next step
class OrderCreateView(View):
    pass


class OrderStatusView(View):
    pass
  • Create core/urls.py to wire the orders URLs into the main project:
from django.contrib import admin
from django.urls import include, path

urlpatterns = [
    path("admin/", admin.site.urls),
    path("api/", include("orders.urls")),
]

What does this URL structure create?

The api/ prefix in core/urls.py combined with orders/ in the app URLs gives you /api/orders/ for creating orders and /api/orders/{id}/status/ for polling status. Django admin remains at /admin/.

  • Save all files (orders/serializers.py, orders/urls.py, orders/views.py, core/urls.py).
  • Bring up the full Docker Compose stack by running:
docker compose up -d --build
  • Run database migrations inside the web container:
docker compose exec web python manage.py migrate
  • Verify the Django service is running by opening http://localhost:8000/admin/ in your browser. You should see the Django admin login page.

Django service not starting?

Run docker compose logs web to check for errors. Common issues:

  • ModuleNotFoundError means a file is in the wrong directory. Check that core/settings.py exists and core/__init__.py imports the Celery app.
  • If migrations fail with a connection error, make sure the postgres service is healthy by running docker compose ps and checking its status.
  • If you see No module named 'orders', verify the orders/ directory has an __init__.py file (even an empty one).

help me debug my Django Docker service startup errors

✔️ Awesome, I've got everything!

Great. Double check you have saved all your files before moving on.

ⓧ I'd like to double check the full code

Here is the complete code for every file you should have at this point.

#!/usr/bin/env python
import os
import sys


def main():
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "core.settings")
    try:
        from django.core.management import execute_from_command_line
    except ImportError as exc:
        raise ImportError(
            "Couldn't import Django. Are you sure it's installed and "
            "available on your PYTHONPATH environment variable? Did you "
            "forget to activate a virtual environment?"
        ) from exc
    execute_from_command_line(sys.argv)


if __name__ == "__main__":
    main()
from .celery import app as celery_app

__all__ = ("celery_app",)
import os

from celery import Celery

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

app = Celery("core")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()
import os

from django.core.asgi import get_asgi_application

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

application = get_asgi_application()
import os
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent

SECRET_KEY = "dev-secret-key-change-in-production"
DEBUG = True
ALLOWED_HOSTS = ["*"]

INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    "rest_framework",
    "django_celery_beat",
    "orders",
]

MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "django.contrib.sessions.middleware.SessionMiddleware",
    "django.middleware.common.CommonMiddleware",
    "django.middleware.csrf.CsrfViewMiddleware",
    "django.contrib.auth.middleware.AuthenticationMiddleware",
    "django.contrib.messages.middleware.MessageMiddleware",
    "django.middleware.clickjacking.XFrameOptionsMiddleware",
]

ROOT_URLCONF = "core.urls"

TEMPLATES = [
    {
        "BACKEND": "django.template.backends.django.DjangoTemplates",
        "DIRS": [],
        "APP_DIRS": True,
        "OPTIONS": {
            "context_processors": [
                "django.template.context_processors.request",
                "django.contrib.auth.context_processors.auth",
                "django.contrib.messages.context_processors.messages",
            ],
        },
    },
]

ASGI_APPLICATION = "core.asgi.application"

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": "orders_db",
        "USER": "orders_user",
        "PASSWORD": "orders_pass",
        "HOST": "postgres",
        "PORT": "5432",
    }
}

CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": os.environ.get("REDIS_URL", "redis://redis:6379/1"),
        "OPTIONS": {
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
        },
    }
}

CELERY_BROKER_URL = os.environ.get("CELERY_BROKER_URL", "amqp://guest:guest@rabbitmq:5672//")
CELERY_RESULT_BACKEND = os.environ.get("CELERY_RESULT_BACKEND", "redis://redis:6379/0")
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERY_TIMEZONE = "UTC"
CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers:DatabaseScheduler"

REST_FRAMEWORK = {
    "DEFAULT_RENDERER_CLASSES": [
        "rest_framework.renderers.JSONRenderer",
        "rest_framework.renderers.BrowsableAPIRenderer",
    ],
}

STATIC_URL = "static/"
DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"
from django.contrib import admin
from django.urls import include, path

urlpatterns = [
    path("admin/", admin.site.urls),
    path("api/", include("orders.urls")),
]
from django.db import models


class Order(models.Model):
    STATUS_CHOICES = [
        ("pending", "Pending"),
        ("processing", "Processing"),
        ("completed", "Completed"),
        ("failed", "Failed"),
    ]

    customer_email = models.EmailField()
    items = models.JSONField()
    total_amount = models.DecimalField(max_digits=10, decimal_places=2)
    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default="pending")
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    failure_reason = models.TextField(blank=True, default="")

    class Meta:
        ordering = ["-created_at"]

    def __str__(self):
        return f"Order {self.id} - {self.status}"
from rest_framework import serializers

from .models import Order


class OrderCreateSerializer(serializers.ModelSerializer):
    class Meta:
        model = Order
        fields = ["customer_email", "items", "total_amount"]


class OrderStatusSerializer(serializers.ModelSerializer):
    class Meta:
        model = Order
        fields = ["id", "status", "created_at", "updated_at", "failure_reason"]
from django.urls import path

from .views import OrderCreateView, OrderStatusView

urlpatterns = [
    path("orders/", OrderCreateView.as_view(), name="order-create"),
    path("orders/<int:order_id>/status/", OrderStatusView.as_view(), name="order-status"),
]

Your Django project is configured and running with Celery, RabbitMQ, and Redis all wired together. Next up, you will replace those placeholder views with async endpoints that create orders and dispatch them into the Celery processing pipeline.

Build Async Order API Endpoints

Your infrastructure is running and the Order model is ready in the database. Now it is time to connect the outside world to your pipeline.

In this step, you will build the two async API endpoints that let clients submit orders and check their status. These views use Django's ASGI async support to handle requests without blocking the server while database operations run.

In this step, get ready to:
  • Create a Celery task stub that the views can dispatch orders to.
  • Build an async POST endpoint that saves orders and publishes them to RabbitMQ.
  • Build an async GET endpoint that reads order status from Redis cache first.
Create the async order creation endpoint

The order creation endpoint does three things: saves the order to PostgreSQL, caches its initial status in Redis, and dispatches a processing task to Celery via RabbitMQ. Since the view imports the task function, you need to create the task file first.

  • Create a new file called orders/tasks.py and add the pipeline orchestrator stub by pasting the following code:
import logging
import random
import time

from celery import shared_task
from django.core.cache import cache

logger = logging.getLogger(__name__)


@shared_task
def process_order_pipeline(order_id):
    # Build a chain of processing tasks for this order
    from celery import chain

    pipeline = chain(
        validate_order.s(order_id),
        process_payment.s(),
        fulfill_order.s(),
    )
    pipeline.apply_async(link_error=handle_order_failure.s(order_id=order_id))

What does this code do?

This file defines the process_order_pipeline task that your view will dispatch. It builds a Celery chain of three processing steps (validate, charge, fulfill) that run in sequence.

The tasks it references (validate_order, process_payment, fulfill_order) do not exist yet. You will implement them in the next step. For now, this stub is enough for your views to import and dispatch.

  • Save orders/tasks.py.

Import errors in your editor?

Your editor may show a warning that validate_order and the other task names are undefined. This is expected. They are referenced inside the function body and will only be resolved at runtime when you add them in the next step.

help me understand why Celery task references can be deferred

Now you can create the view that imports this task and uses it to dispatch orders.

  • Create a new file called orders/views.py and paste the following code:
import json

from asgiref.sync import sync_to_async
from django.core.cache import cache
from django.http import JsonResponse
from django.views import View

from .models import Order
from .serializers import OrderCreateSerializer, OrderStatusSerializer
from .tasks import process_order_pipeline


class OrderCreateView(View):
    async def post(self, request):
        # Parse the incoming JSON request body
        try:
            data = json.loads(request.body)
        except json.JSONDecodeError:
            return JsonResponse({"error": "Invalid JSON"}, status=400)

        # Validate the order data using the DRF serializer
        serializer = OrderCreateSerializer(data=data)
        is_valid = await sync_to_async(lambda: serializer.is_valid())()

        if not is_valid:
            errors = await sync_to_async(lambda: serializer.errors)()
            return JsonResponse({"errors": errors}, status=400)

        # Save the order to PostgreSQL
        order = await sync_to_async(serializer.save)()

        # Cache the initial status in Redis for fast polling
        cache.set(f"order_status:{order.id}", "pending", timeout=3600)

        # Dispatch the processing pipeline to Celery via RabbitMQ
        process_order_pipeline.delay(order.id)

        return JsonResponse(
            {"id": order.id, "status": "pending", "message": "Order submitted for processing"},
            status=201,
        )

What does this code do?

  • async def post makes this an async view. Django serves it through the ASGI server (Uvicorn) without blocking threads.
  • sync_to_async wraps synchronous ORM calls (like serializer.save()) so they can run inside an async function. Django's ORM is not fully async yet, so these wrappers are required.
  • cache.set(f"order_status:{order.id}", "pending", timeout=3600) stores the order's status in Redis. This gives the status-polling endpoint a fast read path without hitting the database.
  • process_order_pipeline.delay(order.id) publishes a message to RabbitMQ. A Celery worker will pick it up and process the order asynchronously.
  • Save orders/views.py.

Seeing a ModuleNotFoundError?

If your editor or container logs show an import error for orders.tasks, make sure you saved orders/tasks.py before creating orders/views.py. The view file imports from tasks, so it must exist first.

help me debug a ModuleNotFoundError for orders.tasks

Create the async order status endpoint

The status endpoint uses a cache-first pattern. It checks Redis for a cached status before falling back to a PostgreSQL query. This keeps polling fast and reduces database load.

  • In orders/views.py, add the following class below OrderCreateView:
class OrderStatusView(View):
    async def get(self, request, order_id):
        # Try Redis cache first (fast path)
        cached_status = cache.get(f"order_status:{order_id}")
        if cached_status:
            return JsonResponse({"id": order_id, "status": cached_status, "source": "cache"})

        # Fall back to PostgreSQL if cache miss
        try:
            order = await sync_to_async(Order.objects.get)(id=order_id)
        except Order.DoesNotExist:
            return JsonResponse({"error": "Order not found"}, status=404)

        serializer = OrderStatusSerializer(order)
        data = await sync_to_async(lambda: serializer.data)()

        # Populate the cache for future requests
        cache.set(f"order_status:{order_id}", order.status, timeout=3600)

        return JsonResponse({**data, "source": "database"})

Why cache-first instead of database-first?

In a production order system, clients poll the status endpoint repeatedly while waiting for their order to complete. Reading from Redis takes microseconds compared to milliseconds for a PostgreSQL query.

The response includes a source field (cache or database) so you can observe which path each request takes during testing.

  • Save orders/views.py.

Indentation error?

Make sure OrderStatusView is at the top level (no indentation) and is not accidentally nested inside OrderCreateView. Both classes should start at column 0.

help me fix an indentation issue in my Django views file

✔️ Awesome, I've got everything!

Great. Double check you have saved both orders/tasks.py and orders/views.py.

ⓧ I'd like to double check the full code

Here is the complete orders/tasks.py:

import logging
import random
import time

from celery import shared_task
from django.core.cache import cache

logger = logging.getLogger(__name__)


@shared_task
def process_order_pipeline(order_id):
    from celery import chain

    pipeline = chain(
        validate_order.s(order_id),
        process_payment.s(),
        fulfill_order.s(),
    )
    pipeline.apply_async(link_error=handle_order_failure.s(order_id=order_id))

Here is the complete orders/views.py:

import json

from asgiref.sync import sync_to_async
from django.core.cache import cache
from django.http import JsonResponse
from django.views import View

from .models import Order
from .serializers import OrderCreateSerializer, OrderStatusSerializer
from .tasks import process_order_pipeline


class OrderCreateView(View):
    async def post(self, request):
        try:
            data = json.loads(request.body)
        except json.JSONDecodeError:
            return JsonResponse({"error": "Invalid JSON"}, status=400)

        serializer = OrderCreateSerializer(data=data)
        is_valid = await sync_to_async(lambda: serializer.is_valid())()

        if not is_valid:
            errors = await sync_to_async(lambda: serializer.errors)()
            return JsonResponse({"errors": errors}, status=400)

        order = await sync_to_async(serializer.save)()

        cache.set(f"order_status:{order.id}", "pending", timeout=3600)

        process_order_pipeline.delay(order.id)

        return JsonResponse(
            {"id": order.id, "status": "pending", "message": "Order submitted for processing"},
            status=201,
        )


class OrderStatusView(View):
    async def get(self, request, order_id):
        cached_status = cache.get(f"order_status:{order_id}")
        if cached_status:
            return JsonResponse({"id": order_id, "status": cached_status, "source": "cache"})

        try:
            order = await sync_to_async(Order.objects.get)(id=order_id)
        except Order.DoesNotExist:
            return JsonResponse({"error": "Order not found"}, status=404)

        serializer = OrderStatusSerializer(order)
        data = await sync_to_async(lambda: serializer.data)()

        cache.set(f"order_status:{order_id}", order.status, timeout=3600)

        return JsonResponse({**data, "source": "database"})
Test the endpoints and verify message flow

With both files saved, Uvicorn's auto-reload will pick up the new views. You also need to restart the Celery worker so it discovers the new task module.

  • Restart the web and worker services so they load the new code by running:
docker compose restart web celery_worker
  • Wait a few seconds for the services to become healthy again.
  • Submit a test order by running this curl command:
curl -X POST http://localhost:8000/api/orders/ \
  -H "Content-Type: application/json" \
  -d '{"customer_email": "test@example.com", "items": [{"name": "Widget", "qty": 2}], "total_amount": "29.99"}'

What should I see?

You should see a JSON response like this:

{"id": 1, "status": "pending", "message": "Order submitted for processing"}

The 201 status code confirms the order was created. The id value may differ if you have previously created orders.

  • Poll the order status by running:
curl http://localhost:8000/api/orders/1/status/

What should I see?

You should see a response showing the cached status:

{"id": 1, "status": "pending", "source": "cache"}

The source: cache field confirms Redis served the response. If you see source: database, that means the cache entry expired or was not set. Check that Redis is healthy.

  • Open your browser and navigate to http://localhost:15672.
  • Log in with username guest and password guest.
  • Click the Queues and Streams tab.
  • Look for the celery queue. You should see message activity confirming that your order was dispatched to RabbitMQ.

Why might the message count be zero?

If the Celery worker consumed the message before you checked, the queue will show 0 messages but the message rate graph will show a spike. The worker may fail to fully process the order (since the individual task steps are not implemented yet), but the message was still successfully published and consumed.

Getting a 500 error or connection refused?

  • Check that all services are running with docker compose ps. All containers should show a healthy status.
  • If the web service is restarting, check its logs with docker compose logs web. Look for import errors or syntax issues in your new files.
  • Make sure orders/urls.py references OrderCreateView and OrderStatusView exactly as shown in the views file.

help me debug a 500 error from my Django async order endpoint

Your async API endpoints are wired up and publishing messages to RabbitMQ. Next up, you will implement the full Celery task chain that validates, charges, and fulfills each order.

Implement the Celery Task Chain Pipeline

Your API can accept orders and dispatch them to Celery, but the actual processing logic does not exist yet.

In this step, you will build the three-step task chain that every order flows through: validation, payment, and fulfillment. You will also wire up automatic retries with exponential backoff and a failure callback that catches permanently failed orders.

In this step, get ready to:
  • Implement the validate, payment, and fulfillment tasks.
  • Wire up error handling with retries and a failure callback.
  • Start the Celery worker, submit test orders, and verify the pipeline in Flower.
Implement the three-step task chain

Each order passes through three stages: validate the order items, process the payment, and fulfill the order. Each stage is a separate Celery task so they can retry independently if something goes wrong.

  • In orders/tasks.py, find the existing process_order_pipeline function.
  • Add the validate_order task above it by copying the following code:
@shared_task(bind=True, max_retries=3, default_retry_delay=5)
def validate_order(self, order_id):
    from .models import Order

    logger.info(f"Validating order {order_id}")
    order = Order.objects.get(id=order_id)

    if not order.items:
        order.status = "failed"
        order.failure_reason = "Order has no items"
        order.save()
        cache.set(f"order_status:{order_id}", "failed", timeout=3600)
        raise ValueError("Order has no items")

    order.status = "processing"
    order.save()
    cache.set(f"order_status:{order_id}", "processing", timeout=3600)

    return order_id

What does this code do?

bind=True gives the task access to self, which you need for retry logic. max_retries=3 caps retry attempts, and default_retry_delay=5 waits 5 seconds between retries.

The task checks whether the order has items. If not, it marks the order as failed in both PostgreSQL and Redis, then raises an exception. If items exist, it transitions the order to "processing" status.

  • Below validate_order, add the process_payment task:
@shared_task(bind=True, max_retries=3, default_retry_delay=10)
def process_payment(self, order_id):
    from .models import Order

    logger.info(f"Processing payment for order {order_id}")

    time.sleep(random.uniform(0.5, 2.0))

    if random.random() < 0.1:
        logger.warning(f"Payment failed for order {order_id}, retrying...")
        raise self.retry(exc=Exception("Payment gateway timeout"))

    return order_id

What does this code do?

This task simulates a real payment gateway. The time.sleep adds a random delay between 0.5 and 2 seconds to mimic network latency.

There is a 10% chance (random.random() < 0.1) the payment fails. When that happens, self.retry() re-queues the task in RabbitMQ. The default_retry_delay=10 gives the payment gateway 10 seconds to recover before retrying.

  • Below process_payment, add the fulfill_order task:
@shared_task(bind=True, max_retries=3, default_retry_delay=5)
def fulfill_order(self, order_id):
    from .models import Order

    logger.info(f"Fulfilling order {order_id}")
    order = Order.objects.get(id=order_id)
    order.status = "completed"
    order.save()
    cache.set(f"order_status:{order_id}", "completed", timeout=3600)

    logger.info(f"Order {order_id} completed successfully")
    return order_id

What does this code do?

The final task in the chain marks the order as "completed" in both PostgreSQL and the Redis cache. This is the happy-path endpoint that every successful order reaches.

  • Save orders/tasks.py.

Seeing import errors?

Make sure you placed all three tasks between the logger = logging.getLogger(__name__) line and the existing process_order_pipeline function. The order matters because process_order_pipeline references validate_order, process_payment, and fulfill_order by name.

help me fix import errors in my Celery tasks file

Wire up error handling and the failure callback

When a task in the chain fails after exhausting all retries, Celery invokes a special error callback. This callback receives the raw request, exception, and traceback directly from the worker.

  • In orders/tasks.py, add the handle_order_failure task below fulfill_order and above the existing process_order_pipeline:
@shared_task
def handle_order_failure(request, exc, traceback, order_id=None):
    from .models import Order

    if order_id:
        logger.error(f"Order {order_id} failed permanently: {exc}")
        order = Order.objects.get(id=order_id)
        order.status = "failed"
        order.failure_reason = str(exc)
        order.save()
        cache.set(f"order_status:{order_id}", "failed", timeout=3600)

How does the failure callback connect to the chain?

The existing process_order_pipeline function already wires this callback via link_error=handle_order_failure.s(order_id=order_id). When any task in the chain raises an unrecoverable exception, Celery calls handle_order_failure directly with the request, exception, and traceback objects.

The order_id keyword argument is passed through the signature so the callback knows which order to mark as permanently failed in both PostgreSQL and Redis.

  • Save orders/tasks.py.

Why import Order inside each task?

Each task uses from .models import Order inside the function body instead of at the top of the file. This avoids circular import issues when Celery autodiscovers tasks before Django has fully loaded its models.

✔️ Awesome, I've got everything!

Great. Double check you have saved orders/tasks.py with all four new tasks in place.

ⓧ I'd like to double check the full code

import logging
import random
import time

from celery import shared_task
from django.core.cache import cache

logger = logging.getLogger(__name__)


@shared_task(bind=True, max_retries=3, default_retry_delay=5)
def validate_order(self, order_id):
    from .models import Order

    logger.info(f"Validating order {order_id}")
    order = Order.objects.get(id=order_id)

    if not order.items:
        order.status = "failed"
        order.failure_reason = "Order has no items"
        order.save()
        cache.set(f"order_status:{order_id}", "failed", timeout=3600)
        raise ValueError("Order has no items")

    order.status = "processing"
    order.save()
    cache.set(f"order_status:{order_id}", "processing", timeout=3600)

    return order_id


@shared_task(bind=True, max_retries=3, default_retry_delay=10)
def process_payment(self, order_id):
    from .models import Order

    logger.info(f"Processing payment for order {order_id}")

    time.sleep(random.uniform(0.5, 2.0))

    if random.random() < 0.1:
        logger.warning(f"Payment failed for order {order_id}, retrying...")
        raise self.retry(exc=Exception("Payment gateway timeout"))

    return order_id


@shared_task(bind=True, max_retries=3, default_retry_delay=5)
def fulfill_order(self, order_id):
    from .models import Order

    logger.info(f"Fulfilling order {order_id}")
    order = Order.objects.get(id=order_id)
    order.status = "completed"
    order.save()
    cache.set(f"order_status:{order_id}", "completed", timeout=3600)

    logger.info(f"Order {order_id} completed successfully")
    return order_id


@shared_task
def handle_order_failure(request, exc, traceback, order_id=None):
    from .models import Order

    if order_id:
        logger.error(f"Order {order_id} failed permanently: {exc}")
        order = Order.objects.get(id=order_id)
        order.status = "failed"
        order.failure_reason = str(exc)
        order.save()
        cache.set(f"order_status:{order_id}", "failed", timeout=3600)


@shared_task
def process_order_pipeline(order_id):
    from celery import chain

    pipeline = chain(
        validate_order.s(order_id),
        process_payment.s(),
        fulfill_order.s(),
    )
    pipeline.apply_async(link_error=handle_order_failure.s(order_id=order_id))
Start the worker and verify the pipeline in Flower

With the tasks written, you need to rebuild the Docker images (the code volume-mounts will pick up changes, but the worker needs a restart to re-import the module).

  • Restart the full stack by running:
docker compose up -d --build

What does this command do?

The --build flag rebuilds the image so the Celery worker picks up the updated orders/tasks.py. All services restart with the new code.

  • Wait about 15 seconds for all health checks to pass, then submit a test order by running:
curl -X POST http://localhost:8000/api/orders/ \
  -H "Content-Type: application/json" \
  -d '{"customer_email": "test@example.com", "items": [{"name": "Widget", "qty": 2}], "total_amount": "29.99"}'
  • You should see a JSON response with "status": "pending" and an order id.
  • Check the order status by running (replace 1 with your returned order ID):
curl http://localhost:8000/api/orders/1/status/
  • After a few seconds, the status should read "completed" with "source": "cache". This confirms the full chain ran: validate, payment, and fulfill.
  • Open Flower at http://localhost:5555 in your browser to see the task chain execution.

In Flower, you should see tasks named orders.tasks.validate_order, orders.tasks.process_payment, and orders.tasks.fulfill_order with a SUCCESS state.

Order stuck in pending?

Check the Celery worker logs for errors by running docker compose logs celery_worker. Look for import errors or connection failures to RabbitMQ.

If you see a connection refused error, make sure the RabbitMQ container is healthy by running docker compose ps. All services should show a "healthy" status.

help me debug why my Celery tasks are not processing orders

Your order processing pipeline is fully operational. Next up, you will add Celery Beat to run periodic maintenance tasks that clean up stale orders automatically.

Add Celery Beat Scheduled Tasks and Run Full Pipeline Test

Your order processing pipeline handles individual orders reliably. But what happens when an order gets stuck in "processing" and never completes? In production, network timeouts, crashed workers, or bugs can leave orders stranded indefinitely.

In this step, you will add an automated cleanup mechanism using Celery Beat. This periodic scheduler runs a task every 60 seconds to find and fail stale orders. Then you will stress-test the full pipeline by submitting 10 orders concurrently and monitoring the results through RabbitMQ Management UI and Flower.

In this step, get ready to:
  • Add a periodic cleanup task that marks stale orders as failed.
  • Run a 10-order concurrent load test across the full pipeline.
  • Monitor system throughput and worker health in real time.
Implement the stale order cleanup task

Orders that stay in "processing" for too long indicate a failure that slipped past the retry mechanism. A periodic task can catch these stragglers and mark them as failed so they do not linger forever.

  • In orders/tasks.py, scroll to the bottom of the file after the process_order_pipeline function.
  • Add the following cleanup_stale_orders task:
@shared_task
def cleanup_stale_orders():
    from datetime import timedelta

    from django.utils import timezone

    from .models import Order

    stale_threshold = timezone.now() - timedelta(minutes=5)
    stale_orders = Order.objects.filter(status="processing", updated_at__lt=stale_threshold)
    count = stale_orders.update(status="failed", failure_reason="Stale order timeout")
    if count:
        logger.info(f"Cleaned up {count} stale orders")

What does this task do?

The task calculates a threshold timestamp 5 minutes in the past. Any order still in "processing" status with an updated_at older than that threshold is considered stale.

It uses a bulk .update() call to mark all stale orders as failed in a single database query. This is efficient even when many orders are stuck. The logger reports the count so you can see cleanup activity in your container logs.

  • Save orders/tasks.py.

You will not see a visible change yet. The next part registers this task with the Beat scheduler.

Register the periodic task with Celery Beat

The django-celery-beat package stores schedules in the database via its DatabaseScheduler. You need to create an IntervalSchedule and a PeriodicTask record so Beat knows to run your cleanup task every 60 seconds.

  • Restart the Celery worker and Beat services so they pick up the new task code by running:
  • Restart the Celery worker and Beat services so they pick up the new task code by running:
docker compose restart celery_worker celery_beat
  • Open a Django shell inside the running web container by running:
docker compose exec web python manage.py shell
  • Create the interval schedule and periodic task by pasting this into the shell:
from django_celery_beat.models import PeriodicTask, IntervalSchedule

# Create a schedule that fires every 60 seconds
schedule, _ = IntervalSchedule.objects.get_or_create(
    every=60,
    period=IntervalSchedule.SECONDS,
)

# Register the cleanup task with that schedule
PeriodicTask.objects.create(
    interval=schedule,
    name="Cleanup stale orders",
    task="orders.tasks.cleanup_stale_orders",
)

What does this do?

IntervalSchedule.objects.get_or_create defines a reusable 60-second interval in the database. PeriodicTask.objects.create registers your cleanup_stale_orders task to run on that interval.

Celery Beat polls the database for schedule changes. Within a few seconds, it will detect the new periodic task and start firing it every 60 seconds.

  • You should see output like <PeriodicTask: Cleanup stale orders: every 60 seconds> confirming the task was registered.
  • Exit the shell by typing exit() and pressing Enter.

Shell command failed?

If you see a database error, make sure your PostgreSQL container is healthy. Run docker compose ps to check service statuses. If the web container is not running, start it with docker compose up -d web.

help me troubleshoot why my Django shell command failed inside Docker

Run a 10-order concurrent load test

Now that the full stack is running with Beat scheduling, it is time to stress-test the pipeline. You will submit 10 orders concurrently and watch them flow through RabbitMQ, the Celery worker, and back to Redis.

  • Submit 10 orders concurrently by running this bash loop:
for i in $(seq 1 10); do
  curl -s -X POST http://localhost:8000/api/orders/ \
    -H "Content-Type: application/json" \
    -d "{\"customer_email\": \"customer${i}@test.com\", \"items\": [{\"name\": \"Item ${i}\", \"qty\": 1}], \"total_amount\": \"${i}9.99\"}" &
done
wait
echo "All 10 orders submitted"

What does this script do?

The & at the end of each curl command sends it to the background, so all 10 requests fire almost simultaneously. The wait command pauses until every background process finishes. This simulates a burst of traffic hitting your API.

  • You should see 10 JSON responses, each containing "status": "pending" and a unique order ID.
  • Wait about 10 seconds for the pipeline to process all orders, then check a few statuses by running:
curl -s http://localhost:8000/api/orders/1/status/ | python3 -m json.tool
curl -s http://localhost:8000/api/orders/5/status/ | python3 -m json.tool
curl -s http://localhost:8000/api/orders/10/status/ | python3 -m json.tool
  • Most orders should show "status": "completed". Some might show "status": "processing" if they are still in-flight, or "status": "failed" if the simulated payment failure triggered.

Payment failures are expected

Your process_payment task has a 10% random failure rate built in. If one or two orders fail, that is the retry mechanism working as designed. Orders that fail all 3 retries get marked as failed by the handle_order_failure callback.

Seeing connection refused errors?

Make sure all services are running with docker compose ps. If the web service restarted after you edited tasks.py, give it a few seconds to come back up. You can also check logs with docker compose logs web --tail=20.

help me debug why my curl requests to localhost:8000 are getting connection refused

Monitor the system via RabbitMQ and Flower

The real value of this architecture is observability. You can watch messages flow through the broker and track task execution across workers in real time.

  • Open the RabbitMQ Management UI at http://localhost:15672 in your browser.
  • Log in with username guest and password guest.
  • Click the Queues and Streams tab to see the celery queue. You should see message rates reflecting your recent load test.

What should I see in RabbitMQ?

The celery queue shows the message publish and delivery rates. After your 10-order burst, you should see a spike in the message rate chart. The queue depth should be near zero if the workers processed everything.

If you see messages sitting in the queue (Ready > 0), your worker might be overloaded or restarting. Check docker compose logs celery_worker --tail=30 for errors.

  • Open Flower at http://localhost:5555 in a new browser tab.
  • Click Tasks in the navigation bar to see the list of executed tasks.
  • You should see orders.tasks.validate_order, orders.tasks.process_payment, and orders.tasks.fulfill_order entries. Most should show a SUCCESS state.
  • Look for orders.tasks.cleanup_stale_orders in the task list. If 60 seconds have passed since you registered the periodic task, Beat should have fired it at least once.

How do I confirm Beat is working?

Check the Celery Beat container logs by running:

docker compose logs celery_beat --tail=20

You should see lines indicating the scheduler is sending the orders.tasks.cleanup_stale_orders task periodically. In Flower, the task will appear with a SUCCESS state and a "received" timestamp that updates every 60 seconds.

Cleanup task not appearing in Flower?

Make sure you restarted celery_beat after adding the task code. Run docker compose restart celery_beat and wait 60 seconds. Also verify the periodic task exists in the database by running docker compose exec web python manage.py shell -c "from django_celery_beat.models import PeriodicTask; print(PeriodicTask.objects.all())".

help me figure out why my Celery Beat periodic task is not firing

✔️ Awesome, I've got everything!

Great. Make sure you have saved orders/tasks.py and that both RabbitMQ Management UI and Flower show your task activity.

ⓧ I'd like to double check the full code

Here is the complete orders/tasks.py with the cleanup task added at the bottom:

import logging
import random
import time

from celery import shared_task
from django.core.cache import cache

logger = logging.getLogger(__name__)


@shared_task(bind=True, max_retries=3, default_retry_delay=5)
def validate_order(self, order_id):
    from .models import Order

    logger.info(f"Validating order {order_id}")
    order = Order.objects.get(id=order_id)

    if not order.items:
        order.status = "failed"
        order.failure_reason = "Order has no items"
        order.save()
        cache.set(f"order_status:{order_id}", "failed", timeout=3600)
        raise ValueError("Order has no items")

    order.status = "processing"
    order.save()
    cache.set(f"order_status:{order_id}", "processing", timeout=3600)

    return order_id


@shared_task(bind=True, max_retries=3, default_retry_delay=10)
def process_payment(self, order_id):
    from .models import Order

    logger.info(f"Processing payment for order {order_id}")

    time.sleep(random.uniform(0.5, 2.0))

    if random.random() < 0.1:
        logger.warning(f"Payment failed for order {order_id}, retrying...")
        raise self.retry(exc=Exception("Payment gateway timeout"))

    return order_id


@shared_task(bind=True, max_retries=3, default_retry_delay=5)
def fulfill_order(self, order_id):
    from .models import Order

    logger.info(f"Fulfilling order {order_id}")
    order = Order.objects.get(id=order_id)
    order.status = "completed"
    order.save()
    cache.set(f"order_status:{order_id}", "completed", timeout=3600)

    logger.info(f"Order {order_id} completed successfully")
    return order_id


@shared_task
def handle_order_failure(request, exc, traceback, order_id=None):
    from .models import Order

    if order_id:
        logger.error(f"Order {order_id} failed permanently: {exc}")
        order = Order.objects.get(id=order_id)
        order.status = "failed"
        order.failure_reason = str(exc)
        order.save()
        cache.set(f"order_status:{order_id}", "failed", timeout=3600)


@shared_task
def process_order_pipeline(order_id):
    from celery import chain

    pipeline = chain(
        validate_order.s(order_id),
        process_payment.s(),
        fulfill_order.s(),
    )
    pipeline.apply_async(link_error=handle_order_failure.s(order_id=order_id))


@shared_task
def cleanup_stale_orders():
    from datetime import timedelta

    from django.utils import timezone

    from .models import Order

    stale_threshold = timezone.now() - timedelta(minutes=5)
    stale_orders = Order.objects.filter(status="processing", updated_at__lt=stale_threshold)
    count = stale_orders.update(status="failed", failure_reason="Stale order timeout")
    if count:
        logger.info(f"Cleaned up {count} stale orders")

Secret mission

Add Dead Letter Queue Handling and Idempotency Keys

Your pipeline processes orders reliably, but what happens when a network glitch causes a client to submit the same order twice? Or when a message permanently fails after all retries? Production systems need safeguards against these exact scenarios.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally via Docker, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Docker containers (postgres, rabbitmq, redis, web, celery_worker, celery_beat, flower).
  • Docker volume (postgres_data).
  • Docker images built from the project Dockerfile.
  • Local project files on your Desktop.

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the order processing pipeline.

  • Your Docker containers will continue running until you manually stop them or restart your machine.
  • RabbitMQ Management UI stays accessible at http://localhost:15672.
  • Flower stays accessible at http://localhost:5555.
  • The Django API stays accessible at http://localhost:8000.

Pause - I'll come back to this later

Shut down running containers to free up memory and CPU. Your data in the PostgreSQL volume is preserved so you can pick up where you left off.

  • Stop all containers by running this command from your project folder:
docker compose down

This stops and removes all containers but keeps the postgres_data volume intact. When you are ready to resume, run docker compose up -d from the same folder.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes your containers, volumes, built images, and local files.

  • Stop all containers, remove volumes, and remove built images by running this command from your project folder:
docker compose down --volumes --rmi local
  • Remove the project folder from your Desktop by running:
cd ~/Desktop
rm -rf order-processing-pipeline

What does this remove?

The --volumes flag removes the postgres_data named volume, deleting all order data stored in PostgreSQL.

The --rmi local flag removes Docker images built from your Dockerfile. Pre-pulled images like postgres:17 and rabbitmq:4.3-management remain in your Docker cache. You can remove those separately from Docker Desktop if you want to reclaim disk space.

Nice Work!

Nice Work!

You just built a production-grade event-driven order processing pipeline from scratch. Your system processes orders through an async API, routes them through a message broker, and handles failures gracefully across multiple services.

You've learned how to:

  • Orchestrate a 6-service Docker Compose stack with Django ASGI, Celery workers, RabbitMQ, Redis, and PostgreSQL communicating through health checks and dependency ordering.
  • Build async REST endpoints that publish orders to a message queue and serve cached status from Redis for fast polling.
  • Implement a Celery task chain with automatic retries, exponential backoff, failure callbacks, and periodic maintenance tasks monitored through Flower.
  • Secret Mission: Added dead letter queue handling and idempotency keys to prevent duplicate order processing.

Ready to quiz yourself?