azure_schema_registry_avro_serializer_serialize_deserialize_quickstart.py

1import os
2from azure.identity import DefaultAzureCredential
3from azure.schemaregistry import SchemaRegistryClient
4from azure.schemaregistry.serializer.avroserializer import AvroSerializer
5
6# Set the environment variables with your own values
7SCHEMAREGISTRY_FULLY_QUALIFIED_NAMESPACE = os.environ['SCHEMAREGISTRY_FULLY_QUALIFIED_NAMESPACE']
8GROUP_NAME = os.environ['SCHEMAREGISTRY_GROUP']
9
10# Create a Schema Registry client
11token_credential = DefaultAzureCredential()
12schema_registry_client = SchemaRegistryClient(
13    fully_qualified_namespace=SCHEMAREGISTRY_FULLY_QUALIFIED_NAMESPACE,
14    credential=token_credential
15)
16
17# Define the Avro schema
18definition = """
19{"namespace": "example.avro",
20 "type": "record",
21 "name": "User",
22 "fields": [
23     {"name": "name", "type": "string"},
24     {"name": "favorite_number",  "type": ["int", "null"]},
25     {"name": "favorite_color", "type": ["string", "null"]}
26 ]
27}"""
28
29# Create an Avro Serializer
30serializer = AvroSerializer(client=schema_registry_client, group_name=GROUP_NAME)
31
32# Data to be serialized
33data = {"name": "Alyssa", "favorite_number": 256, "favorite_color": None}
34
35# Serialize the data
36with serializer:
37    content_type, bytes_data = serializer.serialize(data, schema=definition)
38    print(f"Serialized data: {bytes_data}")
39    print(f"Content type: {content_type}")
40
41    # Deserialize the data
42    deserialized_data = serializer.deserialize(bytes_data, content_type=content_type)
43    print(f"Deserialized data: {deserialized_data}")