kafka-producer-client

Reusable Kafka producer library — install once, use across all your Python projects.
Works in sync scripts, FastAPI, Django, Celery, or any async framework.

Install

# From your private registry / git
pip install py-kafka-producer-client

# Or directly from git
pip install git+https://github.com/your-org/kafka-producer-client.git

Quick Start

1. Sync usage (scripts, Django, Celery workers)

from kafka_producer_client import KafkaProducerClient, KafkaProducerConfig

config = KafkaProducerConfig(
    bootstrap_servers="broker1:9092,broker2:9092",
    default_topic="user-events",
    client_id="my-service",
)

producer = KafkaProducerClient(config)

# Single event
producer.send({"user": "alice", "action": "login"})

# Batch
producer.send_batch([
    {"user": "alice", "action": "login"},
    {"user": "bob", "action": "signup"},
])

# Flush before exit
producer.flush()

2. Async usage (FastAPI)

from fastapi import FastAPI, Depends
from kafka_producer_client import KafkaProducerConfig
from kafka_producer_client.fastapi import create_lifespan, get_producer

config = KafkaProducerConfig(
    bootstrap_servers="broker:9092",
    default_topic="api-events",
)

app = FastAPI(lifespan=create_lifespan(config))

@app.post("/events")
async def post_event(producer=Depends(get_producer)):
    result = await producer.send_async({"event": "click"})
    return result

3. Async usage (standalone asyncio)

import asyncio
from kafka_producer_client import KafkaProducerClient, KafkaProducerConfig

async def main():
    client = KafkaProducerClient(KafkaProducerConfig(
        bootstrap_servers="broker:9092",
    ))
    client.start_background_poll()

    result = await client.send_async(
        {"order_id": 123},
        topic="orders",
        key="order-123",
    )
    print(result)  # {"topic": "orders", "partition": 0, "offset": 42}

    client.close()

asyncio.run(main())

Configuration

All fields on KafkaProducerConfig are optional with sensible defaults:

Passing extra librdkafka config

config = KafkaProducerConfig(
    bootstrap_servers="broker:9092",
    extra={
        "security.protocol": "SASL_SSL",
        "sasl.mechanism": "PLAIN",
        "sasl.username": "key",
        "sasl.password": "secret",
    },
)

API Reference

KafkaProducerClient