Kafka Producer Pattern

Async Kafka producer for FastAPI with mandatory user isolation.

Setup

python
1# services/kafka_producer.py
2import os
3import json
4import logging
5from datetime import datetime, timezone
6from typing import Any
7from aiokafka import AIOKafkaProducer
8
9logger = logging.getLogger(__name__)
10
11_producer: AIOKafkaProducer | None = None
12
13async def get_producer() -> AIOKafkaProducer | None:
14    """Get or create singleton producer. Returns None if Kafka unavailable."""
15    global _producer
16    if _producer is not None:
17        return _producer
18
19    bootstrap_servers = os.getenv("KAFKA_BOOTSTRAP_SERVERS")
20    if not bootstrap_servers:
21        logger.warning("KAFKA_BOOTSTRAP_SERVERS not set, events disabled")
22        return None
23
24    try:
25        _producer = AIOKafkaProducer(
26            bootstrap_servers=bootstrap_servers,
27            value_serializer=lambda v: json.dumps(v).encode("utf-8"),
28        )
29        await _producer.start()
30        logger.info("Kafka producer started")
31        return _producer
32    except Exception as e:
33        logger.error(f"Failed to start Kafka producer: {e}")
34        return None
35
36async def close_producer():
37    """Shutdown producer gracefully."""
38    global _producer
39    if _producer:
40        await _producer.stop()
41        _producer = None

Event Producer Function

python
1# services/kafka_producer.py (continued)
2
3async def produce_event(
4    event_type: str,
5    user_id: int,
6    payload: dict[str, Any],
7    topic: str = "todo-events"
8) -> bool:
9    """
10    Produce event with user isolation. Never crashes endpoint.
11
12    Args:
13        event_type: Event name (e.g., "task-created", "task-completed")
14        user_id: Current user's ID (required for isolation)
15        payload: Event-specific data
16        topic: Kafka topic (default: todo-events)
17
18    Returns:
19        True if sent, False if failed/disabled
20    """
21    producer = await get_producer()
22    if not producer:
23        return False
24
25    event = {
26        "event_type": event_type,
27        "timestamp": datetime.now(timezone.utc).isoformat(),
28        "user_id": user_id,
29        "payload": payload,
30    }
31
32    try:
33        await producer.send_and_wait(topic, event)
34        logger.debug(f"Event produced: {event_type} for user {user_id}")
35        return True
36    except Exception as e:
37        logger.error(f"Failed to produce event: {e}")
38        return False

Endpoint Usage

python
1from services.kafka_producer import produce_event
2
3@router.post("/tasks", response_model=TaskResponse)
4async def create_task(
5    task_data: TaskCreate,
6    current_user: User = Depends(get_current_user)
7):
8    # Create task in database
9    task = await task_service.create(task_data, current_user.id)
10
11    # Produce event (fire-and-forget, won't crash if Kafka down)
12    await produce_event(
13        event_type="task-created",
14        user_id=current_user.id,
15        payload={"task_id": task.id, "title": task.title}
16    )
17
18    return task

Lifespan Integration

python
1# main.py
2from contextlib import asynccontextmanager
3from services.kafka_producer import get_producer, close_producer
4
5@asynccontextmanager
6async def lifespan(app: FastAPI):
7    await get_producer()  # Initialize on startup
8    yield
9    await close_producer()  # Cleanup on shutdown
10
11app = FastAPI(lifespan=lifespan)

Event Schema

All events follow this structure:

json
1{
2  "event_type": "task-created",
3  "timestamp": "2026-02-09T12:00:00+00:00",
4  "user_id": 123,
5  "payload": {
6    "task_id": 456,
7    "title": "Buy groceries"
8  }
9}

Common Event Types

Requirements

aiokafka>=0.10.0

Environment

KAFKA_BOOTSTRAP_SERVERS=localhost:9092