Star Quix Streams

 Install Quix Streams

 Edit on GitHub

Schema Registry

Serializers and deserializers for JSON Schema, Avro, and Protobuf support integration with a Schema Registry.

The current implementation wraps Confluent's serializers and deserializers, which are tightly coupled with the Schema Registry.

To integrate your existing Schema Registry, pass SchemaRegistryClientConfig to your serializers and deserializers. Additional optional configuration can be provided via SchemaRegistrySerializationConfig.

from quixstreams.models import (
    SchemaRegistryClientConfig,
    SchemaRegistrySerializationConfig,
)

schema_registry_client_config = SchemaRegistryClientConfig(
    url='localhost:8081',
    basic_auth_user_info='username:password',
)

# optional
schema_registry_serialization_config = SchemaRegistrySerializationConfig(
    auto_register_schemas=False,
)

Note: For the full list of available options, refer to the Serializers API.

JSON Schema

For both the serializer and deserializer, a schema must be provided.

from quixstreams.models import JSONDeserializer, JSONSerializer

MY_SCHEMA = {
    "title": "MyObject",
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "id": {"type": "number"},
    },
    "required": ["id"],
}

deserializer = JSONDeserializer(
    schema=MY_SCHEMA,
    schema_registry_client_config=schema_registry_client_config,
    schema_registry_serialization_config=schema_registry_serialization_config,
)
serializer = JSONSerializer(
    schema=MY_SCHEMA,
    schema_registry_client_config=schema_registry_client_config,
    schema_registry_serialization_config=schema_registry_serialization_config,
)

Avro

The serializer requires a schema, but the deserializer can automatically fetch the required schema from the Schema Registry.

from quixstreams.models.serialize.avro import AvroDeserializer, AvroSerializer

MY_SCHEMA = {
    "type": "record",
    "name": "testschema",
    "fields": [
        {"name": "name", "type": "string"},
        {"name": "id", "type": "int", "default": 0},
    ],
}

deserializer = AvroDeserializer(
    schema_registry_client_config=schema_registry_client_config,
    schema_registry_serialization_config=schema_registry_serialization_config,
)
serializer = AvroSerializer(
    schema=MY_SCHEMA,
    schema_registry_client_config=schema_registry_client_config,
    schema_registry_serialization_config=schema_registry_serialization_config,
)

Protobuf

For both the serializer and deserializer, msg_type must be provided.

from quixstreams.models.serialize.protobuf import ProtobufDeserializer, ProtobufSerializer

from my_input_models_pb2 import InputProto
from my_output_models_pb2 import OutputProto

deserializer = ProtobufDeserializer(
    msg_type=InputProto,
    schema_registry_client_config=schema_registry_client_config,
    schema_registry_serialization_config=schema_registry_serialization_config,
)
serializer = ProtobufSerializer(
    msg_type=OutputProto,
    schema_registry_client_config=schema_registry_client_config,
    schema_registry_serialization_config=schema_registry_serialization_config,
)

See the Serialization and Deserialization page to learn more about how to integrate the serializer and deserializer with your application.