Building event-driven systems can be super powerful, especially when you’re looking for something scalable and easy to maintain. If you mix FastAPI, a modern Python web framework, with Apache Kafka, an awesome event streaming platform, you’re basically setting yourself up to handle tons of data and keep things communicating in real-time between various parts of your system. Let’s dive into how we can make this happen.
What’s Event-Driven Architecture Anyway?
Event-driven architecture is pretty much about events. Think of them like little notifications saying, “Hey, something happened!” One part of the system shoots off these events, and other parts pick them up. This setup allows the whole system to be loosely coupled, meaning you can add new stuff without breaking all the existing bits. It’s like adding new apps on your phone without crashing the whole OS.
Kicking Things Off with FastAPI and Kafka
FastAPI is dope for building APIs in Python because it supports asynchronous programming, making it lightning fast. Kafka, on the other hand, is the heavyweight champ for event streaming.
First things first, we need to get FastAPI installed along with its friends. You can use Poetry, which is a nifty dependency manager for Python:
poetry install fastapi uvicorn
Next up, Kafka. We’ll run Kafka locally using Docker Compose to keep things simple. Here’s a sample docker-compose.yml file to get you going:
version: '3'
services:
zookeeper:
image: confluentinc/cp-zookeeper:latest
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
kafka:
image: confluentinc/cp-kafka:latest
depends_on:
- zookeeper
ports:
- "9092:9092"
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
Launch Kafka like so:
docker-compose up -d
Producing and Consuming Events with Kafka
So, in this event-driven system, events are fired into Kafka topics and other components can grab them. Producing an event is straightforward:
from confluent_kafka import Producer
producer = Producer({
'bootstrap.servers': 'localhost:9092',
'client.id': 'sample-producer'
})
def produce_event(topic, message):
producer.produce(topic, value=message)
producer.poll(0)
produce_event('my_topic', 'Hello, Kafka!')
Consuming those events is just as easy:
from confluent_kafka import Consumer
consumer = Consumer({
'bootstrap.servers': 'localhost:9092',
'group.id': 'my_group',
'auto.offset.reset': 'earliest'
})
def consume_events(topic):
consumer.subscribe([topic])
while True:
message = consumer.poll(1.0)
if message is None:
continue
elif message.error():
print(f"Error: {message.error()}")
else:
print(f"Received message: {message.value().decode('utf-8')}")
consume_events('my_topic')
Marrying FastAPI with Kafka
FastAPI can generate endpoints that push events straight into Kafka topics. Check this out:
from fastapi import FastAPI
from confluent_kafka import Producer
app = FastAPI()
producer = Producer({
'bootstrap.servers': 'localhost:9092',
'client.id': 'fastapi-producer'
})
@app.post("/produce_event")
async def produce_event(topic: str, message: str):
producer.produce(topic, value=message)
producer.poll(0)
return {"message": "Event produced successfully"}
And to consume events, you might want to run a background task:
from fastapi import FastAPI, BackgroundTasks
from confluent_kafka import Consumer
app = FastAPI()
consumer = Consumer({
'bootstrap.servers': 'localhost:9092',
'group.id': 'fastapi_group',
'auto.offset.reset': 'earliest'
})
def consume_events(topic):
consumer.subscribe([topic])
while True:
message = consumer.poll(1.0)
if message is None:
continue
elif message.error():
print(f"Error: {message.error()}")
else:
print(f"Received message: {message.value().decode('utf-8')}")
@app.on_event("startup")
async def startup_event():
import threading
threading.Thread(target=consume_events, args=("my_topic",)).start()
@app.get("/")
async def read_root():
return {"message": "Welcome to the FastAPI and Kafka integration"}
Bringing in Domain-Driven Design (DDD)
For more complex systems, Domain-Driven Design (DDD) is golden. It helps keep the code organized around the business logic. Here’s a sneaky peek into how you can structure your project with DDD:
- api: FastAPI endpoints.
- domain: Business entities and events.
- infrastructure: Repositories and Kafka integration.
- logic: Handlers for commands and events.
- settings: Configuration.
- tests: Your test cases.
A Real-World Layout
Imagine your project directory looking something like this:
project/
├── api/
│ ├── main.py
│ └── routes/
│ ├── __init__.py
│ └── events.py
├── domain/
│ ├── __init__.py
│ ├── entities.py
│ └── events.py
├── infrastructure/
│ ├── __init__.py
│ ├── kafka.py
│ └── repositories.py
├── logic/
│ ├── __init__.py
│ ├── command_handlers.py
│ └── event_handlers.py
├── settings.py
└── tests/
├── __init__.py
└── test_events.py
Dealing with Commands and Events
Commands run through a mediator that calls the relevant handlers. These handlers usually mess with databases and can even produce events for Kafka. Here’s a taste:
from typing import Dict
class Mediator:
def __init__(self, command_handlers: Dict[str, callable], event_handlers: Dict[str, callable]):
self.command_handlers = command_handlers
self.event_handlers = event_handlers
def handle_command(self, command):
handler = self.command_handlers.get(type(command).__name__)
if handler:
handler(command)
def handle_event(self, event):
handler = self.event_handlers.get(type(event).__name__)
if handler:
handler(event)
class CreateEventCommand:
def __init__(self, event_data):
self.event_data = event_data
class CreateEventHandler:
def __init__(self, kafka_producer):
self.kafka_producer = kafka_producer
def __call__(self, event):
self.kafka_producer.produce('events', value=event.event_data)
mediator = Mediator(
command_handlers={
'CreateEventCommand': lambda command: print(f"Creating event: {command.event_data}"),
},
event_handlers={
'CreateEvent': CreateEventHandler(producer),
}
)
command = CreateEventCommand('Hello, Kafka!')
mediator.handle_command(command)
Real-World Example: Ticket Booking System
Picture this, a ticket booking system. Whenever a new conference is set up, the organizer sends a “Create conference” API request via FastAPI. This gets pushed to the “Conferences” topic in Kafka. The data is then processed and sent to clients in real-time. Here’s a simple version:
from fastapi import FastAPI
from confluent_kafka import Producer
app = FastAPI()
producer = Producer({
'bootstrap.servers': 'localhost:9092',
'client.id': 'ticket-booking-producer'
})
@app.post("/create_conference")
async def create_conference(data: str):
producer.produce('conferences', value=data)
producer.poll(0)
return {"message": "Conference created successfully"}
@app.post("/book_ticket")
async def book_ticket(data: str):
producer.produce('bookings', value=data)
producer.poll(0)
return {"message": "Ticket booked successfully"}
Wrapping Up
Combining FastAPI and Kafka can give you a robust and scalable system to handle real-time data like a pro. By tapping into the strengths of both, you get a setup that’s loosely coupled, easy to maintain, and can handle a ton of data. Whether you’re building a chat app, a ticket booking system, or anything else, FastAPI and Kafka offer an epic foundation for your architecture.
