Producer

KafkaProducer takes a KafkaConfig and produces batches. Use it as a context manager so it flushes and closes on exit.

from pyrula.kafka import KafkaConfig, KafkaProducer

with KafkaProducer(KafkaConfig(bootstrap_servers="localhost:9092")) as producer:
    receipt = producer.produce_many_bytes("events", values=[b"a", b"b"], keys=[b"k1", b"k2"])
    receipt.get().unwrap_or_raise()

values and keys are parallel lists. There’s a path per payload type: produce_many_bytes, produce_many_json, and produce_many_avro. See Formats. For a single message, producer.produce("events", b"v", key=b"k") takes the same receipt; the batch paths are what you reach for at volume.

Delivery

produce_many_* returns a ProduceReceipt without blocking. Decide when to wait:

receipt = producer.produce_many_bytes("events", values=batch)
# ... do other work while it's in flight ...
report = receipt.get().unwrap_or_raise()   # blocks until the broker acks

receipt.get() and receipt.wait() block for the batch’s delivery report. To drain everything that’s queued, call producer.flush(timeout_ms=...). producer.purge() drops anything still queued without sending.

Delivery guarantees

Set these on the config:

• acks="all" (the default) waits for the in-sync replicas.
• enable_idempotence=True makes retries safe, so a retried batch won’t duplicate.
• compression, linger_ms, and batch_size_bytes trade latency for throughput.

For exactly-once across a consume/produce cycle, use Transactions.

Async

AsyncKafkaProducer has the same surface with await and async with:

from pyrula.kafka import AsyncKafkaProducer

async with AsyncKafkaProducer(config) as producer:
    await producer.produce_many_bytes("events", values=batch)
    await producer.flush()