azure_schema_registry_avro_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 up the SchemaRegistryClient
7endpoint = os.environ["AZURE_SCHEMA_REGISTRY_ENDPOINT"]
8credential = DefaultAzureCredential()
9client = SchemaRegistryClient(endpoint, credential)
10
11# Set up the AvroSerializer
12group_name = "your-schema-group"
13serializer = AvroSerializer(client=client, group_name=group_name)
14
15# Define the schema and the data to be serialized
16schema = """
17{"namespace": "example.avro",
18 "type": "record",
19 "name": "User",
20 "fields": [
21     {"name": "name", "type": "string"},
22     {"name": "favorite_number",  "type": ["int", "null"]},
23     {"name": "favorite_color", "type": ["string", "null"]}
24 ]
25}"""
26data = {"name": "Alyssa", "favorite_number": 256, "favorite_color": None}
27
28# Serialize the data
29content_type, serialized_data = serializer.serialize(data, schema=schema)
30
31# Deserialize the data
32deserialized_data = serializer.deserialize(serialized_data, content_type=content_type)
33
34print(f"Deserialized data: {deserialized_data}")