Event-Driven Order Processing API

Build an async order API with FastAPI, Kafka, and Redis running locally.

Introduction

30 Second Summary

Think about ordering food through a delivery app. You tap "Place Order" and get an instant confirmation, but the restaurant hasn't even started cooking yet.

In this project, you will build an event-driven order processing API that immediately confirms orders while processing them asynchronously in the background using Kafka and Redis.

What You'll Build

A FastAPI REST API that publishes order events to Apache Kafka, processes them asynchronously with a background consumer, and tracks real-time order status in Redis.

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

  • A running FastAPI application with endpoints to place orders and check their status in real time.
  • A Kafka producer and background consumer that publish and process order events asynchronously.
  • A Docker Compose stack running Kafka (KRaft mode) and Redis locally with zero cloud costs.
  • Secret Mission: Add a dead letter queue for failed orders and an endpoint to retry them.

Are there any prerequisites?

You should have basic Python knowledge and some familiarity with REST APIs. No prior experience with Kafka, Redis, or event-driven architecture is needed.

Set Up Docker Services

Your Python environment is ready with all the packages installed. Now you need the infrastructure those packages will talk to.

This project uses two backend services: Apache Kafka for streaming order events and Redis for caching order status. Instead of installing them directly on your machine, you will run them in Docker containers orchestrated by Docker Compose.

In this step, get ready to:
  • Create a Docker Compose configuration for Kafka and Redis.
  • Pin your project dependencies in a requirements file.
  • Start both services and verify they are healthy.
Create the Docker Compose file

Docker Compose lets you define multiple services in a single YAML file and start them all with one command. Your project needs Kafka running in KRaft mode (which removes the need for a separate ZooKeeper process) and Redis for fast key-value lookups.

  • Create a file called docker-compose.yml in your project folder by running:
touch docker-compose.yml
  • Open docker-compose.yml in your editor and paste the following configuration:
services:
  kafka:
    image: apache/kafka:4.2.0
    hostname: kafka
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
    volumes:
      - kafka-data:/tmp/kraft-combined-logs
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 || exit 1"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 30s

  redis:
    image: redis:latest
    container_name: redis
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

volumes:
  kafka-data:

What does this configuration do?

The kafka service uses the official apache/kafka:4.2.0 image running in KRaft mode. KRaft means Kafka handles its own metadata internally, so you do not need a separate ZooKeeper container. The environment variables configure a single-node cluster that listens on port 9092 for client connections and port 9093 for internal controller communication.

The redis service runs on port 6379 with a simple redis-cli ping healthcheck. Both services include healthchecks so Docker can report when each container is ready to accept connections.

The kafka-data volume persists Kafka's log data between container restarts, so you do not lose messages if you stop and restart the stack.

  • Save docker-compose.yml.

YAML formatting issues?

YAML is sensitive to indentation. Make sure you use spaces (not tabs) and that each nested level is indented by exactly 2 spaces. If your editor auto-converts spaces to tabs, check its settings.

Help me fix YAML indentation errors in my docker-compose.yml file

Pin your dependencies in a requirements file

A requirements.txt file records the exact package versions your project uses. This makes it easy for anyone (including future you) to recreate the same environment.

  • Create a file called requirements.txt and paste the following:
fastapi[standard]==0.136.3
aiokafka==0.14.0
redis==8.0.0

What are these packages?

  • fastapi[standard] is the web framework for building your API, bundled with uvicorn for serving it.
  • aiokafka is an async Kafka client built for Python's asyncio event loop. It integrates naturally with FastAPI's async architecture.
  • redis is the official Python client for Redis, with built-in async support via redis.asyncio.
  • Save requirements.txt.
  • Confirm both files exist by running:
ls

You should see docker-compose.yml and requirements.txt listed in the output.

Start and verify the Docker services

With the configuration in place, you can now bring up both Kafka and Redis with a single command.

  • Start the Docker Compose stack in detached mode by running:
docker compose up -d

What does this command do?

The -d flag runs the containers in the background (detached mode) so your terminal stays free. Docker will pull the Kafka and Redis images the first time you run this, which may take a minute or two depending on your internet speed.

Kafka takes about 30 seconds to fully initialize its KRaft metadata before it reports healthy. Wait about 30-40 seconds, then check the status.

  • Verify both containers are healthy by running:
docker compose ps

You should see both kafka and redis listed with a STATUS showing Up and (healthy).

Services not showing as healthy?

If Kafka shows (health: starting), wait another 30 seconds and run docker compose ps again. Kafka's start_period: 30s setting means Docker will not mark it unhealthy until at least 30 seconds have passed.

If a container shows as exited or unhealthy after a full minute, check its logs by running docker compose logs kafka or docker compose logs redis. The most common issue is port 9092 or 6379 already being used by another process on your machine.

Help me troubleshoot my Docker Compose Kafka and Redis services

✔️ Awesome, I've got everything!

Great. Make sure you have saved docker-compose.yml and requirements.txt, and that both containers show healthy.

ⓧ I'd like to double check the full code

Here is the complete docker-compose.yml:

services:
  kafka:
    image: apache/kafka:4.2.0
    hostname: kafka
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
    volumes:
      - kafka-data:/tmp/kraft-combined-logs
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 || exit 1"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 30s

  redis:
    image: redis:latest
    container_name: redis
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

volumes:
  kafka-data:

And here is the complete requirements.txt:

fastapi[standard]==0.136.3
aiokafka==0.14.0
redis==8.0.0

Your infrastructure is running. Next up, you will build the FastAPI application that connects to Kafka and Redis to process orders.

Build the FastAPI App with Kafka Producer

Your infrastructure is running. Kafka and Redis are healthy inside Docker, ready to receive messages and cache data.

Now you need the application layer that ties everything together. In this step, you'll build a FastAPI app that publishes order events to Kafka and stores order status in Redis for instant lookups.

In this step, get ready to:
  • Create main.py with a FastAPI lifespan that manages your Kafka producer and Redis connection.
  • Build a POST endpoint that publishes order events to Kafka and caches their status.
  • Build a GET endpoint that retrieves order status from Redis.
Create the FastAPI app with lifespan

The lifespan pattern in FastAPI lets you start external connections (like Kafka and Redis) when the server boots and cleanly shut them down when it stops. This replaces the older startup/shutdown event hooks.

  • Create a new file called main.py in your project directory.
  • Add the imports, configuration constants, data model, and global connection variables by pasting this code:
import asyncio
import json
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone

import redis.asyncio as aioredis
from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel


# Connection settings for Kafka and Redis
KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"
KAFKA_TOPIC = "orders"
REDIS_URL = "redis://localhost:6379"


# Define the shape of incoming order requests
class OrderRequest(BaseModel):
    product: str
    quantity: int
    customer_name: str


# Global references to shared connections
redis_client: aioredis.Redis = None
producer: AIOKafkaProducer = None

What does this code do?

The top section imports async libraries for Kafka (aiokafka) and Redis (redis.asyncio), plus FastAPI and Pydantic for the web layer.

The three constants define where Kafka and Redis live on your machine. OrderRequest is a Pydantic model that validates incoming JSON automatically. The two global variables (redis_client and producer) will hold your live connections once the server starts.

You won't see a change yet. The next chunk adds the consumer function that processes orders from Kafka.

  • Add the order consumer function below the globals by pasting this code at the bottom of main.py:

async def consume_orders():
    consumer = AIOKafkaConsumer(
        KAFKA_TOPIC,
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        group_id="order-processor",
        value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    )
    await consumer.start()
    try:
        async for msg in consumer:
            order_data = msg.value
            order_id = order_data["order_id"]

            # Simulate processing time
            await asyncio.sleep(1)

            # Update status in Redis
            await redis_client.hset(
                f"order:{order_id}",
                mapping={"status": "completed", "processed_at": datetime.now(timezone.utc).isoformat()},
            )
    finally:
        await consumer.stop()

What does this code do?

This function subscribes to the orders Kafka topic and loops over incoming messages. For each order, it waits one second (simulating real processing work), then flips the order's status to completed in Redis.

The value_deserializer automatically converts raw Kafka bytes back into a Python dictionary. The try/finally ensures the consumer disconnects cleanly even if an error occurs.

You won't see a change yet. The next chunk creates the lifespan that starts the producer, Redis, and this consumer together.

  • Add the lifespan context manager and app instantiation below consume_orders() by pasting this code at the bottom of main.py:

@asynccontextmanager
async def lifespan(app: FastAPI):
    global redis_client, producer

    # Connect to Redis and start the Kafka producer
    redis_client = aioredis.from_url(REDIS_URL, decode_responses=True)
    producer = AIOKafkaProducer(
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        value_serializer=lambda v: json.dumps(v).encode("utf-8"),
    )
    await producer.start()

    # Launch the consumer as a background task
    consumer_task = asyncio.create_task(consume_orders())

    yield

    # Cleanup on shutdown
    consumer_task.cancel()
    await producer.stop()
    await redis_client.aclose()


app = FastAPI(lifespan=lifespan)

What does this code do?

Everything before yield runs when the server starts. It connects to Redis, starts the Kafka producer, and launches the consumer as a background task using asyncio.create_task(). This means the consumer runs in the same event loop as your API.

Everything after yield runs on shutdown. It cancels the consumer, stops the producer, and closes the Redis connection. No resource leaks.

Why a lifespan instead of startup events?

The older @app.on_event("startup") pattern is deprecated in FastAPI. The lifespan context manager gives you a single function that handles both startup and shutdown, making resource management cleaner and harder to get wrong.

  • Save main.py.
  • Start the development server by running:
fastapi dev main.py
  • You should see output indicating the server is running at http://127.0.0.1:8000. The app starts even though there are no routes yet.
  • Press Ctrl+C in your terminal to stop the server for now. You'll restart it after adding the endpoints.

Server won't start?

  • If you see a connection error mentioning Kafka or Redis, make sure your Docker containers are still running. Run docker compose ps to check.
  • If you see a ModuleNotFoundError, make sure your virtual environment is activated before running the command.
  • If you see a syntax error, double check your indentation matches the code above exactly.

Help me debug why my FastAPI server won't start with Kafka and Redis connections.

Create the POST /orders endpoint

This endpoint is the entry point for the entire event-driven flow. When a client places an order, the API immediately returns a response with the order ID. The actual processing happens asynchronously through Kafka.

  • Add the POST /orders endpoint below app = FastAPI(lifespan=lifespan) by pasting this code at the bottom of main.py:

@app.post("/orders")
async def create_order(order: OrderRequest):
    order_id = str(uuid.uuid4())
    now = datetime.now(timezone.utc).isoformat()

    # Build the event payload
    order_event = {
        "order_id": order_id,
        "product": order.product,
        "quantity": order.quantity,
        "customer_name": order.customer_name,
        "created_at": now,
    }

    # Publish to Kafka
    await producer.send_and_wait(KAFKA_TOPIC, value=order_event)

    # Cache initial status in Redis
    await redis_client.hset(
        f"order:{order_id}",
        mapping={
            "order_id": order_id,
            "product": order.product,
            "quantity": str(order.quantity),
            "customer_name": order.customer_name,
            "status": "pending",
            "created_at": now,
        },
    )
    await redis_client.zadd("orders:timeline", {order_id: datetime.now(timezone.utc).timestamp()})

    return {"order_id": order_id, "status": "pending"}

What does this code do?

  • Generates a unique order_id using UUID4 so every order has a distinct identifier.
  • Publishes the full order event to Kafka using send_and_wait(), which guarantees the message is written before continuing.
  • Stores the order details as a Redis hash at the key order:{order_id} with an initial status of pending.
  • Adds the order ID to a sorted set (orders:timeline) scored by timestamp, so you can query orders chronologically later.
  • Returns immediately with the order ID and pending status. The client never waits for processing to finish.
  • Save main.py and start the server again by running:
fastapi dev main.py
  • Open a second terminal window and test the endpoint by running:
curl -X POST http://127.0.0.1:8000/orders -H "Content-Type: application/json" -d '{"product": "keyboard", "quantity": 1, "customer_name": "Alice"}'
  • You should see a JSON response containing an order_id (a long UUID string) and "status": "pending".

Seeing a connection refused error?

  • Make sure the first terminal is still running fastapi dev main.py. The server must be active to accept requests.
  • Verify your Docker containers are still running with docker compose ps. If Kafka or Redis stopped, run docker compose up -d to restart them.

Help me debug my POST /orders endpoint that returns a connection error.

Create the GET /orders/{order_id} endpoint

Clients need a way to check whether their order has been processed. This endpoint reads the order hash directly from Redis, giving instant lookups without touching Kafka.

  • Press Ctrl+C in the terminal running your server to stop it.
  • Add the GET /orders/{order_id} endpoint below the create_order function by pasting this code at the bottom of main.py:

@app.get("/orders/{order_id}")
async def get_order(order_id: str):
    # Look up the full order hash from Redis
    order_data = await redis_client.hgetall(f"order:{order_id}")
    if not order_data:
        raise HTTPException(status_code=404, detail="Order not found")
    return order_data

What does this code do?

The hgetall() call retrieves every field from the Redis hash stored at order:{order_id}. If the order doesn't exist, Redis returns an empty dictionary, and the endpoint raises a 404 error.

  • Save main.py and restart the server by running:
fastapi dev main.py
  • In your second terminal, place a new order by running:
curl -X POST http://127.0.0.1:8000/orders -H "Content-Type: application/json" -d '{"product": "monitor", "quantity": 2, "customer_name": "Bob"}'
  • Copy the order_id value from the response.
  • Check the order status by running this command, replacing the ID with your actual order ID:
curl http://127.0.0.1:8000/orders/[[ORDER_ID="your-order-id-from-the-previous-response"]]
  • You should see the full order details including "status": "completed". The background consumer already processed the order and updated Redis while you were reading the response.

Catch it in pending state

The consumer takes about 1 second to process each order. If you GET the order immediately after POSTing (within that 1 second window), you'll see "status": "pending". Try it quickly to see the async flow in action.

Getting a 404 Order not found?

  • Make sure you copied the full UUID from the POST response. It should look like a1b2c3d4-e5f6-7890-abcd-ef1234567890.
  • If you restarted the server between POSTing and GETting, your Redis data from the previous run is still there. The order ID from the old POST still works.

Help me figure out why my GET /orders endpoint returns 404.

✔️ Awesome, I've got everything!

Great. Double check you've saved main.py and your server is running without errors.

ⓧ I'd like to double check the full code

Here's the complete main.py file as it should look at the end of this step:

import asyncio
import json
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone

import redis.asyncio as aioredis
from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel


KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"
KAFKA_TOPIC = "orders"
REDIS_URL = "redis://localhost:6379"


class OrderRequest(BaseModel):
    product: str
    quantity: int
    customer_name: str


redis_client: aioredis.Redis = None
producer: AIOKafkaProducer = None


async def consume_orders():
    consumer = AIOKafkaConsumer(
        KAFKA_TOPIC,
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        group_id="order-processor",
        value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    )
    await consumer.start()
    try:
        async for msg in consumer:
            order_data = msg.value
            order_id = order_data["order_id"]

            # Simulate processing time
            await asyncio.sleep(1)

            # Update status in Redis
            await redis_client.hset(
                f"order:{order_id}",
                mapping={"status": "completed", "processed_at": datetime.now(timezone.utc).isoformat()},
            )
    finally:
        await consumer.stop()


@asynccontextmanager
async def lifespan(app: FastAPI):
    global redis_client, producer

    redis_client = aioredis.from_url(REDIS_URL, decode_responses=True)
    producer = AIOKafkaProducer(
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        value_serializer=lambda v: json.dumps(v).encode("utf-8"),
    )
    await producer.start()

    consumer_task = asyncio.create_task(consume_orders())

    yield

    consumer_task.cancel()
    await producer.stop()
    await redis_client.aclose()


app = FastAPI(lifespan=lifespan)


@app.post("/orders")
async def create_order(order: OrderRequest):
    order_id = str(uuid.uuid4())
    now = datetime.now(timezone.utc).isoformat()

    order_event = {
        "order_id": order_id,
        "product": order.product,
        "quantity": order.quantity,
        "customer_name": order.customer_name,
        "created_at": now,
    }

    # Publish to Kafka
    await producer.send_and_wait(KAFKA_TOPIC, value=order_event)

    # Cache initial status in Redis
    await redis_client.hset(
        f"order:{order_id}",
        mapping={
            "order_id": order_id,
            "product": order.product,
            "quantity": str(order.quantity),
            "customer_name": order.customer_name,
            "status": "pending",
            "created_at": now,
        },
    )
    await redis_client.zadd("orders:timeline", {order_id: datetime.now(timezone.utc).timestamp()})

    return {"order_id": order_id, "status": "pending"}


@app.get("/orders/{order_id}")
async def get_order(order_id: str):
    order_data = await redis_client.hgetall(f"order:{order_id}")
    if not order_data:
        raise HTTPException(status_code=404, detail="Order not found")
    return order_data

Your event-driven API is live. Orders flow from the REST endpoint, through Kafka, and back into Redis with an updated status. Next up, you'll add an order history endpoint that queries Redis for chronological order listings.

Build the Background Kafka Consumer

Your API can place orders and cache their status in Redis. But right now, those orders stay pending forever because nothing is processing them on the other side of Kafka.

In this step, you'll understand how the background consumer picks up order events from Kafka and updates their status in Redis. Then you'll test the full end-to-end flow to watch an order transition from pending to completed in real time.

In this step, get ready to:
  • Review the Kafka consumer function that processes order events.
  • Understand how the consumer runs as a background task inside FastAPI's lifespan.
  • Test the full event-driven flow from placing an order to seeing it completed.
Review the Kafka consumer logic

The consume_orders() function in main.py is the other half of your event-driven system. While the POST /orders endpoint publishes events to Kafka, this consumer subscribes to the same topic and processes them.

  • In main.py, scroll up and find the consume_orders() function.
  • Read through it to identify what happens when a message arrives from the orders topic.
async def consume_orders():
    consumer = AIOKafkaConsumer(
        KAFKA_TOPIC,
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        group_id="order-processor",
        value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    )
    await consumer.start()
    try:
        async for msg in consumer:
            order_data = msg.value
            order_id = order_data["order_id"]

            # Simulate processing time
            await asyncio.sleep(1)

            # Update status in Redis
            await redis_client.hset(
                f"order:{order_id}",
                mapping={"status": "completed", "processed_at": datetime.now(timezone.utc).isoformat()},
            )
    finally:
        await consumer.stop()

What does this function do?

This function creates an AIOKafkaConsumer that subscribes to the orders topic. The group_id identifies this consumer as part of the order-processor group, which means Kafka tracks which messages it has already read.

The value_deserializer converts raw bytes from Kafka back into a Python dictionary. For each message, it extracts the order_id, waits one second to simulate processing work, then updates the order status to completed in Redis.

The finally block ensures the consumer connection closes cleanly even if an error occurs.

What is a consumer group?

A consumer group lets multiple consumer instances share the work of processing a topic. Kafka tracks which messages each group has read, so a message is only processed once per group.

If you scaled this app to three instances, Kafka would distribute the order events across all three consumers in the same group automatically.

Understand the background task integration

A consumer that runs forever needs to coexist with your API server in the same process. FastAPI's lifespan context manager is where long-running background tasks get launched and cleaned up.

  • In main.py, find the lifespan function and locate this line:
    consumer_task = asyncio.create_task(consume_orders())

Why asyncio.create_task() instead of await?

If you used await consume_orders(), the lifespan function would block forever waiting for the consumer's infinite loop to finish. Your API server would never start.

Using asyncio.create_task() schedules the consumer to run in the background on the same event loop. Both the API and the consumer share the event loop cooperatively, each yielding control during await calls.

  • Now look at the cleanup code after yield:
    consumer_task.cancel()
    await producer.stop()
    await redis_client.aclose()

What happens at shutdown?

When you press Ctrl+C to stop the server, FastAPI resumes the lifespan after the yield. The cleanup cancels the consumer task, stops the Kafka producer, and closes the Redis connection. This prevents resource leaks.

Test the full event-driven flow

Now you'll see the entire system in action. You'll place an order, watch it start as pending, then query it again after the consumer processes it to see it flip to completed.

  • Start the FastAPI development server by running this command in your terminal:
fastapi dev main.py

What does this command do?

The fastapi dev command starts your app in development mode with auto-reload enabled. It listens on http://localhost:8000 by default.

Server won't start?

Make sure your Docker Compose stack is still running. The app needs Kafka on port 9092 and Redis on port 6379. Run docker compose ps to check both containers show a healthy status.

If you see a ModuleNotFoundError, make sure your virtual environment is activated and you've installed all dependencies from requirements.txt.

Help me troubleshoot why my FastAPI server won't start. I'm building an event-driven order processing system with Kafka and Redis.

  • Open a second terminal window. Place a test order by running this command:
curl -X POST http://localhost:8000/orders -H "Content-Type: application/json" -d '{"product": "Keyboard", "quantity": 1, "customer_name": "Alice"}'

You should see a JSON response with an order_id and a status of pending.

  • Copy the order_id value from the response. Immediately check the order status by running:
curl http://localhost:8000/orders/[[ORDER_ID="paste-your-order-id-here"]]

If you check quickly enough (within one second), you'll see "status": "pending". The consumer hasn't finished processing yet.

  • Wait two seconds, then run the same GET request again:
curl http://localhost:8000/orders/[[ORDER_ID="paste-your-order-id-here"]]

This time the status should show "status": "completed" along with a processed_at timestamp. The consumer processed the event from Kafka and updated Redis.

What just happened?

You witnessed the core event-driven architecture pattern in action. The POST endpoint returned immediately with pending status. The client didn't wait for processing to finish.

Behind the scenes, the Kafka producer published the order event to the orders topic. The background consumer picked it up, simulated one second of work, then updated the status in Redis to completed. This decoupling is how production systems handle high throughput without making users wait.

Status stays pending?

Check the terminal where your FastAPI server is running. Look for error messages from the consumer. A common issue is that Kafka hasn't finished creating the orders topic yet. Wait 10 seconds and try placing another order.

If the consumer keeps failing silently, verify Kafka is healthy by running docker compose ps and confirming the kafka container status says healthy.

Help me debug why my Kafka consumer isn't processing orders

Your event-driven system is processing orders asynchronously. Next up, you'll add order history and use Redis data patterns to build a timeline of all orders.

Add Order History and Redis Data Patterns

Your event-driven system is humming along. Orders flow through Kafka, get processed in the background, and their status updates land in Redis in real time.

Right now, you can check a single order's status if you know its ID. In this step, you will add a listing endpoint that leverages the Redis data patterns already in your code to serve order history with instant lookups.

In this step, get ready to:
  • Understand the Redis hash and sorted set patterns powering your order system.
  • Create a GET /orders endpoint for browsing order history.
  • Test the complete three-endpoint API end to end.
Understand your Redis data patterns

Before adding the new endpoint, take a moment to understand the two Redis data structures your create_order function already uses.

What are hashes and sorted sets in Redis?

Your code uses two complementary Redis data structures to store order data.

  • Hash (order:{order_id}): stores the full order details as field-value pairs. Think of it like a Python dictionary living inside Redis. Each order gets its own hash with fields like product, quantity, status, and created_at.
  • Sorted set (orders:timeline): indexes order IDs by their creation timestamp. The timestamp is the "score" that Redis uses to keep members ordered. This lets you retrieve the most recent orders without scanning every hash.

Why two structures instead of one?

Hashes store rich data but have no built-in ordering. Sorted sets provide ordering but only store member strings and scores. By combining them, you get the best of both: the sorted set answers "which orders are newest?" and the hashes answer "what are the full details of each order?"

This is a common Redis pattern in production systems. It avoids expensive full-key scans and keeps queries fast regardless of how many orders exist.

  • In main.py, find the create_order function and locate these two Redis calls near the bottom:
    # Cache initial status in Redis
    await redis_client.hset(
        f"order:{order_id}",
        mapping={
            "order_id": order_id,
            "product": order.product,
            "quantity": str(order.quantity),
            "customer_name": order.customer_name,
            "status": "pending",
            "created_at": now,
        },
    )
    await redis_client.zadd("orders:timeline", {order_id: datetime.now(timezone.utc).timestamp()})

What do these lines do?

  • hset creates (or updates) a Redis hash at the key order:{order_id}. The mapping dict becomes the hash's field-value pairs.
  • zadd adds the order ID to the orders:timeline sorted set with the current Unix timestamp as its score. Higher scores (more recent timestamps) sort to the top when retrieved in reverse order.

These two structures are the foundation for the listing endpoint you will build next.

Create the order listing endpoint

Now you will add a GET /orders endpoint that queries the sorted set for recent order IDs, then fetches each order's full details from its hash.

  • In main.py, find the get_order function. Add the following new endpoint below it:
@app.get("/orders")
async def list_orders(limit: int = 10):
    order_ids = await redis_client.zrevrange("orders:timeline", 0, limit - 1)
    orders = []
    for oid in order_ids:
        order_data = await redis_client.hgetall(f"order:{oid}")
        if order_data:
            orders.append(order_data)
    return orders

What does this code do?

  • zrevrange("orders:timeline", 0, limit - 1) retrieves order IDs from the sorted set in reverse score order (newest first). The 0 and limit - 1 arguments define the index range to return.
  • The limit query parameter defaults to 10, so callers can request fewer or more results by passing ?limit=5 in the URL.
  • For each order ID, hgetall fetches every field from that order's hash. The if order_data check handles the edge case where a hash was deleted but its sorted set entry remains.
  • Save main.py.
  • Your dev server should auto-reload. If you stopped it earlier, restart it by running:
fastapi dev main.py
  • Place a couple of test orders by running these commands in a new terminal:
curl -X POST http://localhost:8000/orders -H "Content-Type: application/json" -d '{"product": "Mechanical Keyboard", "quantity": 1, "customer_name": "Alice"}'

curl -X POST http://localhost:8000/orders -H "Content-Type: application/json" -d '{"product": "USB-C Monitor", "quantity": 2, "customer_name": "Bob"}'
  • Wait about 2 seconds for the consumer to process both orders, then fetch the order list by running:
curl http://localhost:8000/orders

You should see a JSON array containing both orders with their full details and a completed status.

Getting an empty array?

  • Make sure your Docker Compose stack is still running. Run docker compose ps and confirm both kafka and redis show as healthy.
  • Check that your FastAPI server reloaded after saving. Look for a "Reloading..." message in the terminal where it is running.
  • If you restarted Docker Compose between steps, previous order data was lost. Place new orders using the curl commands above and try again.

Help me debug my GET /orders endpoint

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and your server is running.

ⓧ I'd like to double check the full code

Here is the complete main.py at this point:

import asyncio
import json
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone

import redis.asyncio as aioredis
from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel


KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"
KAFKA_TOPIC = "orders"
REDIS_URL = "redis://localhost:6379"


class OrderRequest(BaseModel):
    product: str
    quantity: int
    customer_name: str


redis_client: aioredis.Redis = None
producer: AIOKafkaProducer = None


async def consume_orders():
    consumer = AIOKafkaConsumer(
        KAFKA_TOPIC,
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        group_id="order-processor",
        value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    )
    await consumer.start()
    try:
        async for msg in consumer:
            order_data = msg.value
            order_id = order_data["order_id"]

            # Simulate processing time
            await asyncio.sleep(1)

            # Update status in Redis
            await redis_client.hset(
                f"order:{order_id}",
                mapping={"status": "completed", "processed_at": datetime.now(timezone.utc).isoformat()},
            )
    finally:
        await consumer.stop()


@asynccontextmanager
async def lifespan(app: FastAPI):
    global redis_client, producer

    redis_client = aioredis.from_url(REDIS_URL, decode_responses=True)
    producer = AIOKafkaProducer(
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        value_serializer=lambda v: json.dumps(v).encode("utf-8"),
    )
    await producer.start()

    consumer_task = asyncio.create_task(consume_orders())

    yield

    consumer_task.cancel()
    await producer.stop()
    await redis_client.aclose()


app = FastAPI(lifespan=lifespan)


@app.post("/orders")
async def create_order(order: OrderRequest):
    order_id = str(uuid.uuid4())
    now = datetime.now(timezone.utc).isoformat()

    order_event = {
        "order_id": order_id,
        "product": order.product,
        "quantity": order.quantity,
        "customer_name": order.customer_name,
        "created_at": now,
    }

    # Publish to Kafka
    await producer.send_and_wait(KAFKA_TOPIC, value=order_event)

    # Cache initial status in Redis
    await redis_client.hset(
        f"order:{order_id}",
        mapping={
            "order_id": order_id,
            "product": order.product,
            "quantity": str(order.quantity),
            "customer_name": order.customer_name,
            "status": "pending",
            "created_at": now,
        },
    )
    await redis_client.zadd("orders:timeline", {order_id: datetime.now(timezone.utc).timestamp()})

    return {"order_id": order_id, "status": "pending"}


@app.get("/orders/{order_id}")
async def get_order(order_id: str):
    order_data = await redis_client.hgetall(f"order:{order_id}")
    if not order_data:
        raise HTTPException(status_code=404, detail="Order not found")
    return order_data


@app.get("/orders")
async def list_orders(limit: int = 10):
    order_ids = await redis_client.zrevrange("orders:timeline", 0, limit - 1)
    orders = []
    for oid in order_ids:
        order_data = await redis_client.hgetall(f"order:{oid}")
        if order_data:
            orders.append(order_data)
    return orders

Your event-driven order processing system is now feature-complete. You have three endpoints working together: one to place orders, one to check individual order status, and one to browse order history.

Secret mission

Dead Letter Queue for Failed Orders

In production, event processing fails. Network timeouts, invalid data, downstream service outages. What happens to those events? You need a dead letter queue to catch failures and a retry mechanism to recover from them.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally using Docker containers, 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 (Kafka and Redis) running via Docker Compose.
  • Docker volume (kafka-data) storing Kafka logs.
  • Python virtual environment with installed packages (fastapi, aiokafka, redis).
  • Project files (main.py, docker-compose.yml, requirements.txt).

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the event-driven system.

  • Your Kafka and Redis containers continue running in the background.
  • You can restart the FastAPI server anytime with fastapi dev main.py.
  • Docker containers use system memory while running. Stop them with the Pause instructions if you need to free resources.

Pause - I'll come back to this later

Shut down running processes to free up memory while keeping all your files and Docker images intact.

  • Stop the FastAPI server by pressing Ctrl+C in the terminal where it is running.
  • Stop the Docker containers (while keeping the volume data) by running this command from your project folder:
docker compose down
  • When you are ready to resume, navigate back to your project folder and run:
docker compose up -d
  • Your Kafka data persists in the kafka-data volume, so previous messages are still available.

Delete - I don't want to use this again

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

  • Stop the FastAPI server by pressing Ctrl+C in the terminal where it is running.
  • Stop and remove containers, networks, and the Kafka data volume by running this command from your project folder:
docker compose down -v

What does this command do?

The -v flag removes named volumes declared in the Compose file. Without it, the kafka-data volume would persist on your system even after the containers are gone.

  • Deactivate your Python virtual environment by running:
deactivate
  • Navigate to your Desktop and delete the project folder by running:
cd ~/Desktop
rm -rf order-processing

Folder name may differ

Replace order-processing with whatever you named your project folder in Step 1. You can run ls to check the folder name before deleting.

  • On Windows, use this command instead:
cd ~\Desktop
Remove-Item -Recurse -Force order-processing
  • Confirm the folder is gone by running ls (macOS/Linux) or dir (Windows). You should no longer see your project folder listed.

Nice Work!

Nice Work!

You just built a production-grade event-driven architecture running entirely on your local machine. Orders flow through Kafka, get processed asynchronously, and have their status tracked in real time through Redis.

You've learned how to:

  • Build a FastAPI REST API that publishes order events to Kafka using an async producer and returns responses immediately without waiting for processing.
  • Run a background Kafka consumer inside the FastAPI lifespan that processes events asynchronously and updates order status in Redis in real time.
  • Orchestrate Kafka (KRaft mode) and Redis with Docker Compose for a fully local, zero-cost development stack.
  • Added a dead letter queue with retry capabilities so failed orders are never lost and can be re-processed on demand.

Ready to quiz yourself?