你当前正在访问 Microsoft Azure Global Edition 技术文档网站。如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

适用于 Python 的 Azure 架构注册表客户端库 - 版本 1.2.0

项目
04/05/2023

Azure 架构注册表是由 Azure 事件中心托管的架构存储库服务，提供架构存储、版本控制和管理。序列化程序利用注册表来减小有效负载大小，同时使用架构标识符而不是完整架构描述有效负载结构。

免责声明

对 Python 2.7 的 Azure SDK Python 包支持已于 2022 年 1 月 1 日结束。有关详细信息和问题，请参阅 https://github.com/Azure/azure-sdk-for-python/issues/20691

入门

安装包

使用 pip 安装适用于 Python 的 Azure 架构注册表客户端库：

pip install azure-schemaregistry

先决条件：

若要使用此包，必须具有：

Azure 订阅 - 创建免费帐户
Azure 架构注册表 - 下面是使用 Azure 门户创建架构注册表组的快速入门指南。
Python 3.7 或更高版本 - 安装 Python

验证客户端

与架构注册表的交互从 SchemaRegistryClient 类的实例开始。客户端构造函数采用完全限定的命名空间和 Azure Active Directory 凭据：

架构注册表实例的完全限定命名空间应采用以下格式： <yournamespace>.servicebus.windows.net。
应将实现 TokenCredential 协议的 AAD 凭据传递给构造函数。 azure-identity 包中提供了协议的实现TokenCredential。若要使用提供的 azure-identity凭据类型，请使用 pip 安装适用于 Python 的 Azure 标识客户端库：

pip install azure-identity

此外，若要使用异步 API，必须先安装异步传输，例如 aiohttp：

pip install aiohttp

使用 azure-identity 库创建客户端：

from azure.schemaregistry import SchemaRegistryClient
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
# Namespace should be similar to: '<your-eventhub-namespace>.servicebus.windows.net/'
fully_qualified_namespace = '<< FULLY QUALIFIED NAMESPACE OF THE SCHEMA REGISTRY >>'
schema_registry_client = SchemaRegistryClient(fully_qualified_namespace, credential)

关键概念

架构：架构是数据的组织或结构。可在此处找到更多详细信息。
架构组：基于业务条件的类似架构的逻辑组，可以保存架构的多个版本。可在此处找到更多详细信息。
SchemaRegistryClient： SchemaRegistryClient 提供用于在架构注册表中存储和检索架构的 API。

示例

以下部分提供了几个代码片段，涵盖了一些最常见的架构注册表任务，包括：

注册架构
按 ID 获取架构
按版本获取架构
获取架构的 ID

注册架构

使用 SchemaRegistryClient.register_schema 方法注册架构。

import os

from azure.identity import DefaultAzureCredential
from azure.schemaregistry import SchemaRegistryClient

token_credential = DefaultAzureCredential()
fully_qualified_namespace = os.environ['SCHEMA_REGISTRY_FULLY_QUALIFIED_NAMESPACE']
group_name = os.environ['SCHEMA_REGISTRY_GROUP']
name = "your-schema-name"
format = "Avro"
definition = """
{"namespace": "example.avro",
 "type": "record",
 "name": "User",
 "fields": [
     {"name": "name", "type": "string"},
     {"name": "favorite_number",  "type": ["int", "null"]},
     {"name": "favorite_color", "type": ["string", "null"]}
 ]
}
"""

schema_registry_client = SchemaRegistryClient(fully_qualified_namespace=fully_qualified_namespace, credential=token_credential)
with schema_registry_client:
    schema_properties = schema_registry_client.register_schema(group_name, name, definition, format)
    id = schema_properties.id

按 ID 获取架构

按架构 ID 获取架构定义及其属性。

import os

from azure.identity import DefaultAzureCredential
from azure.schemaregistry import SchemaRegistryClient

token_credential = DefaultAzureCredential()
fully_qualified_namespace = os.environ['SCHEMA_REGISTRY_FULLY_QUALIFIED_NAMESPACE']
schema_id = 'your-schema-id'

schema_registry_client = SchemaRegistryClient(fully_qualified_namespace=fully_qualified_namespace, credential=token_credential)
with schema_registry_client:
    schema = schema_registry_client.get_schema(schema_id)
    definition = schema.definition
    properties = schema.properties

按版本获取架构

按架构版本获取架构定义及其属性。

import os

from azure.identity import DefaultAzureCredential
from azure.schemaregistry import SchemaRegistryClient

token_credential = DefaultAzureCredential()
fully_qualified_namespace = os.environ['SCHEMA_REGISTRY_FULLY_QUALIFIED_NAMESPACE']
group_name = os.environ["SCHEMAREGISTRY_GROUP"]
name = "your-schema-name"
version = int("<your schema version>")

schema_registry_client = SchemaRegistryClient(fully_qualified_namespace=fully_qualified_namespace, credential=token_credential)
with schema_registry_client:
    schema = schema_registry_client.get_schema(group_name=group_name, name=name, version=version)
    definition = schema.definition
    properties = schema.properties

获取架构的 ID

按架构定义及其属性获取架构的架构 ID。

import os

from azure.identity import DefaultAzureCredential
from azure.schemaregistry import SchemaRegistryClient

token_credential = DefaultAzureCredential()
fully_qualified_namespace = os.environ['SCHEMA_REGISTRY_FULLY_QUALIFIED_NAMESPACE']
group_name = os.environ['SCHEMA_REGISTRY_GROUP']
name = "your-schema-name"
format = "Avro"
definition = """
{"namespace": "example.avro",
 "type": "record",
 "name": "User",
 "fields": [
     {"name": "name", "type": "string"},
     {"name": "favorite_number",  "type": ["int", "null"]},
     {"name": "favorite_color", "type": ["string", "null"]}
 ]
}
"""

schema_registry_client = SchemaRegistryClient(fully_qualified_namespace=fully_qualified_namespace, credential=token_credential)
with schema_registry_client:
    schema_properties = schema_registry_client.register_schema(group_name, name, definition, format)
    id = schema_properties.id

疑难解答

常规

架构注册表客户端会引发 Azure Core 中定义的异常。

日志记录

此库使用标准日志记录库进行日志记录。有关 HTTP 会话 (URL、标头等的基本信息，) 在 INFO 级别记录。

可以使用参数在客户端 logging_enable 上启用详细的调试级别日志记录，包括请求/响应正文和未处理标头：

import sys
import logging
from azure.schemaregistry import SchemaRegistryClient
from azure.identity import DefaultAzureCredential

# Create a logger for the SDK
logger = logging.getLogger('azure.schemaregistry')
logger.setLevel(logging.DEBUG)

# Configure a console output
handler = logging.StreamHandler(stream=sys.stdout)
logger.addHandler(handler)

credential = DefaultAzureCredential()
# This client will log detailed information about its HTTP sessions, at DEBUG level
schema_registry_client = SchemaRegistryClient("your_fully_qualified_namespace", credential, logging_enable=True)

同样，即使没有为客户端启用详细日志记录，logging_enable 也可以为单个操作启用：

schema_registry_client.get_schema(schema_id, logging_enable=True)

后续步骤

贡献

本项目欢迎贡献和建议。大多数贡献要求你同意贡献者许可协议 (CLA)，并声明你有权（并且确实有权）授予我们使用你的贡献的权利。有关详细信息，请访问 https://cla.microsoft.com 。

提交拉取请求时，CLA 机器人将自动确定你是否需要提供 CLA，并相应地修饰 PR（例如标签、注释）。直接按机器人提供的说明操作。只需使用 CLA 对所有存储库执行一次这样的操作。

此项目采用了 Microsoft 开放源代码行为准则。有关详细信息，请参阅行为准则常见问题解答，或如果有任何其他问题或意见，请与联系。

通过