1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
|
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
"""
Example to show basic usage of schema registry asynchronously:
- register a schema
- get schema by id
- get schema id
"""
import asyncio
import os
from azure.identity.aio import ClientSecretCredential
from azure.schemaregistry.aio import SchemaRegistryClient
from azure.schemaregistry import SerializationType
TENANT_ID = os.environ['SCHEMA_REGISTRY_AZURE_TENANT_ID']
CLIENT_ID = os.environ['SCHEMA_REGISTRY_AZURE_CLIENT_ID']
CLIENT_SECRET = os.environ['SCHEMA_REGISTRY_AZURE_CLIENT_SECRET']
SCHEMA_REGISTRY_ENDPOINT = os.environ['SCHEMA_REGISTRY_ENDPOINT']
SCHEMA_GROUP = os.environ['SCHEMA_REGISTRY_GROUP']
SCHEMA_NAME = 'your-schema-name'
SERIALIZATION_TYPE = SerializationType.AVRO
SCHEMA_STRING = """
{"namespace": "example.avro",
"type": "record",
"name": "User",
"fields": [
{"name": "name", "type": "string"},
{"name": "favorite_number", "type": ["int", "null"]},
{"name": "favorite_color", "type": ["string", "null"]}
]
}"""
async def register_schema(client, schema_group, schema_name, serialization_type, schema_string):
print("Registering schema...")
schema_properties = await client.register_schema(schema_group, schema_name, serialization_type, schema_string)
print("Schema registered, returned schema id is {}".format(schema_properties.schema_id))
print("Schema properties are {}".format(schema_properties))
return schema_properties.schema_id
async def get_schema_by_id(client, schema_id):
print("Getting schema by id...")
schema = await client.get_schema(schema_id)
print("The schema string of schema id: {} string is {}".format(schema_id, schema.schema_content))
print("Schema properties are {}".format(schema_id))
return schema.schema_content
async def get_schema_id(client, schema_group, schema_name, serialization_type, schema_string):
print("Getting schema id...")
schema_properties = await client.get_schema_id(schema_group, schema_name, serialization_type, schema_string)
print("The schema id is: {}".format(schema_properties.schema_id))
print("Schema properties are {}".format(schema_properties))
return schema_properties.schema_id
async def main():
token_credential = ClientSecretCredential(
tenant_id=TENANT_ID,
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET
)
schema_registry_client = SchemaRegistryClient(endpoint=SCHEMA_REGISTRY_ENDPOINT, credential=token_credential)
async with token_credential, schema_registry_client:
schema_id = await register_schema(schema_registry_client, SCHEMA_GROUP, SCHEMA_NAME, SERIALIZATION_TYPE, SCHEMA_STRING)
schema_str = await get_schema_by_id(schema_registry_client, schema_id)
schema_id = await get_schema_id(schema_registry_client, SCHEMA_GROUP, SCHEMA_NAME, SERIALIZATION_TYPE, SCHEMA_STRING)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
|
