FastAPI + Kafka — Build Real-Time Event Systems

Introduction

Kafka is the spine of most modern event-driven systems. Pairing it with FastAPI gives you HTTP APIs in front of a real-time event stream — perfect for order pipelines, notifications, audit logs and analytics. This tutorial uses aiokafka and walks through producer, consumer, schemas and operational pitfalls.

For the Java equivalent, see Spring Boot + Kafka — Build a Real-Time Messaging System.

Key takeaways

• Use aiokafka to avoid blocking the FastAPI event loop.
• Producers should set acks=all and enable_idempotence=True in production.
• Consumers must handle at-least-once semantics — design idempotent handlers.
• Validate every event with Pydantic before processing.
• Always commit offsets after successful processing, not before.

Setup

pip install aiokafka pydantic

Producer wired into FastAPI's lifespan

```python
# app/events/producer.py
from contextlib import asynccontextmanager
from aiokafka import AIOKafkaProducer
from fastapi import FastAPI
import orjson

producer: AIOKafkaProducer | None = None

@asynccontextmanager async def lifespan(app: FastAPI): global producer producer = AIOKafkaProducer( bootstrap_servers="kafka:9092", enable_idempotence=True, acks="all", compression_type="zstd", ) await producer.start() try: yield finally: await producer.stop()

app = FastAPI(lifespan=lifespan)

async def publish(topic: str, key: str, value: dict): await producer.send_and_wait(topic, orjson.dumps(value), key=key.encode()) ```

Endpoint that emits an event

@app.post("/orders")
async def create_order(payload: OrderCreate):
    order = await svc.create(payload)
    await publish("orders.created", key=str(order.id), value=order.model_dump(mode="json"))
    return order

Consumer service

```python
# consumer/main.py
import asyncio, orjson
from aiokafka import AIOKafkaConsumer
from app.schemas.events import OrderCreated

async def run(): consumer = AIOKafkaConsumer( "orders.created", bootstrap_servers="kafka:9092", group_id="notifications", enable_auto_commit=False, auto_offset_reset="earliest", ) await consumer.start() try: async for msg in consumer: event = OrderCreated.model_validate_json(msg.value) try: await handle(event) await consumer.commit() except Exception: # leave offset uncommitted → message will be re-delivered raise finally: await consumer.stop()

asyncio.run(run()) ```

Schemas

Define event contracts as Pydantic models in a shared package versioned alongside producer + consumer. For polyglot teams, Avro + Schema Registry is the right next step.

Production best practices

• Tune partitions for parallelism — a topic with one partition cannot scale beyond one consumer.
• Keep handlers idempotent: use a dedupe table keyed by event id.
• Set max.poll.interval.ms higher than your slowest handler.
• Use a dead-letter topic for poison messages; never drop silently.

Common mistakes

• Auto-committing offsets — you'll lose messages on crash mid-processing.
• Doing heavy work inside the producer endpoint — push to Kafka, return 202, process async.
• Sharing a consumer group across unrelated services — they'll steal each other's messages.

Tools used in this tutorial

The setup below uses the following tools. Versions matter less than the role each one plays, so swap freely as long as you keep the responsibilities the same.

• Python 3.12+
• FastAPI
• Uvicorn / Gunicorn
• Pydantic v2
• SQLAlchemy 2.x
• Docker
• Kubernetes

Real-world production context

In production, a Python backend like this rarely runs as a single uvicorn process on a laptop. Teams build the app into a container image, push it to a registry, and run it behind a managed load balancer with autoscaling, health checks and centralized logging. Async FastAPI services typically use multiple Uvicorn workers managed by Gunicorn, a managed Postgres for state, Redis for cache and rate-limit counters, and an object store for uploads. The reason cloud hosting becomes a hard requirement is operational: zero-downtime rollouts, log aggregation, metrics, and TLS at the edge are not optional once real users depend on the API.

Recommended tools & deployment options

Once the tutorial works on your machine, the next question is *where do I run this for real?* These are the platforms most Python backend teams reach for:

• DigitalOcean — the simplest path from a working FastAPI container to a public URL. App Platform deploys directly from a Dockerfile, managed Postgres and Redis are one click away, and pricing is predictable. A common way to deploy the setup in this tutorial is using a cloud provider like DigitalOcean when you want to ship quickly without learning a full cloud SDK.
• AWS — the default for enterprise workloads. ECS Fargate or EKS run containers without you managing servers, RDS handles Postgres, and CloudWatch covers logs and metrics.
• Docker — the packaging format every modern deploy target understands. Build once, run the same image locally, in CI and in production.
• Kubernetes (managed: EKS, DOKS, GKE) — the right choice once you have more than a handful of services, need rolling updates, autoscaling and policy-driven networking.

A VPS or managed cloud service is required to run this architecture end-to-end — uvicorn --reload is for development, not for serving traffic.

FAQ

Kafka vs Redis Streams vs RabbitMQ? Kafka for high throughput + replay. Redis Streams for simple, single-cluster pipelines. RabbitMQ for classic task queues.

Next steps & related tutorials

Keep the momentum going with the next tutorial in this learning path:

• Next: FastAPI Microservices Architecture
• Next: Deploy FastAPI Applications to Kubernetes
• Next: Java equivalent: Spring Boot + Kafka