Formats

The producer has a path per payload type. Pick the one that matches what’s on the wire.

producer.produce_many_bytes("events", values=[b"raw", b"bytes"])
producer.produce_many_json("events", values=[{"id": 1}, {"id": 2}])
producer.produce_many_avro("events", values=[{"id": 1}], serde=my_avro_serde)

produce_many_bytes ships your bytes untouched. produce_many_json encodes each value as JSON. produce_many_avro takes an AvroSerde instance and encodes the whole batch in Rust against the schema it carries.

Schema Registry

For the Confluent wire format ([0x00][schema_id][payload]), use the registry serdes from the core package. They register the schema, look up schema ids, and frame the payload so Confluent consumers can read it.

from pyrula import AvroRegistrySerde

serde = AvroRegistrySerde.for_topic("http://localhost:8081", "events", SCHEMA)
framed = serde.serialize({"id": 1, "name": "alice"})   # Ok(bytes) with framing
producer.produce_many_bytes("events", values=[framed.get_or_else(b"")])

JsonSchemaRegistrySerde and ProtobufRegistrySerde cover the other two formats. They handle TLS and mTLS to the registry. Full details and the per-format caveats are on the Serde page.

Decoding on consume

Records come back as bytes in record.value. Decode with the matching serde:

for record in batch.records.to_list():
    obj = serde.deserialize(record.value)   # Ok(dict) or Err