Skip to content
Pyrula

Quickstart

pip install pyrula-kafka. The API tracks confluent-kafka, with two differences: it’s batch-first, and calls return an Either instead of raising. You check is_ok() rather than wrapping every call in try/except.

These examples assume an events topic. A local broker auto-creates it on first use; managed Kafka usually does not, so create it first (see Admin).

Produce

Section titled “Produce”
from pyrula.kafka import KafkaConfig, KafkaProducer


config = KafkaConfig(bootstrap_servers="localhost:9092")


with KafkaProducer(config) as producer:
    receipt = producer.produce_many_bytes("events", values=[b"a", b"b", b"c"])
    report = receipt.get().unwrap_or_raise()   # blocks until the broker acks the batch

produce_many_bytes queues a batch and hands back a ProduceReceipt. Call .get() (or .wait()) when you want to block for delivery.

Consume

Section titled “Consume”
from pyrula.kafka import KafkaConsumerConfig, KafkaConsumer


config = KafkaConsumerConfig(
    bootstrap_servers="localhost:9092",
    topics=["events"],
    group_id="my-group",
)


with KafkaConsumer(config) as consumer:
    while True:
        result = consumer.poll_batch(max_records=500, timeout_ms=1000)
        if result.is_err():
            raise RuntimeError(result.error)
        batch = result.value
        if len(batch) == 0:
            continue          # early polls can be empty while the group joins
        for record in batch:
            print(record.partition, record.offset, record.value)
        consumer.commit(batch).unwrap_or_raise()
        break

poll_batch returns Ok(batch) or Err(error). A batch is iterable and sized: for record in batch and len(batch) work directly, and it carries the offsets to commit. Each record has topic, partition, offset, key, value (bytes), headers, and timestamp_ms. The first few polls after subscribing can return an empty batch while the consumer group joins, so poll in a loop.

Why batch-first

Section titled “Why batch-first”

One poll returns up to max_records records, and one produce call enqueues a whole list. The per-message overhead that dominates a Python produce/consume loop is paid once per batch instead of once per record, which is where the throughput comes from.

Next

Section titled “Next”