confluent_avro 1.5.0

An Avro SerDe implementation that integrates with the confluent

confluent-avro

An Avro SerDe implementation that integrates with the confluent schema registry and serializes and deserializes data according to the defined confluent wire format

Installation

pip install confluent-avro

Usage:

Check examples for a full working demo.

Consumer App Example:

from kafka import KafkaConsumer

from confluent_avro.schema_registry import SchemaRegistry
from confluent_avro.schema_registry.auth import RegistryHTTPBasicAuth
from confluent_avro.serde import AvroKeyValueSerde

KAFKA_TOPIC = "confluent_avro-example-topic"

registry_client = SchemaRegistry(
    "https://myschemaregistry.com",
    RegistryHTTPBasicAuth("username", "password"),
    headers={"Content-Type": "application/vnd.schemaregistry.v1+json"},
)
avroSerde = AvroKeyValueSerde(registry_client, KAFKA_TOPIC)

consumer = KafkaConsumer(
    KAFKA_TOPIC,
    group_id="random_group_id",
    bootstrap_servers=["localhost:9092",]
)

for msg in consumer:
    v = avroSerde.value.deserialize(msg.value)
    k = avroSerde.key.deserialize(msg.key)
    print(msg.offset, msg.partition, k, v)

Producer App Example:

from kafka import KafkaProducer

from confluent_avro.schema_registry import SchemaRegistry
from confluent_avro.schema_registry.auth import RegistryHTTPBasicAuth
from confluent_avro.serde import AvroKeyValueSerde

KAFKA_TOPIC = "confluent_avro-example-topic"

registry_client = SchemaRegistry(
    "https://myschemaregistry.com",
    RegistryHTTPBasicAuth("username", "password"),
    headers={"Content-Type": "application/vnd.schemaregistry.v1+json"},
)

avroSerde = AvroKeyValueSerde(registry_client, KAFKA_TOPIC)

producer = KafkaProducer(bootstrap_servers=["localhost:9092"])
producer.send(
    KAFKA_TOPIC,
    key=avroSerde.key.serialize({...}, key_schema),
    value=avroSerde.value.serialize({...}, value_schema),
)