Python 用 Azure Schema Registry クライアント ライブラリ - バージョン 1.2.0

  • [アーティクル]

Azure Schema Registry は、スキーマ ストレージ、バージョン管理、管理を提供する、Azure Event Hubsによってホストされるスキーマ リポジトリ サービスです。 レジストリはシリアライザーによって利用され、完全なスキーマではなくスキーマ識別子を使用してペイロード構造を記述しながら、ペイロード サイズを小さくします。

ソースコード | パッケージ (PyPi) | API リファレンス ドキュメント | サンプル | Changelog

免責事項

Python 2.7 の Azure SDK Python パッケージのサポートは、2022 年 1 月 1 日に終了しました。 詳細と質問については、https://github.com/Azure/azure-sdk-for-python/issues/20691 を参照してください

作業の開始

パッケージをインストールする

pip を使用して Python 用 Azure Schema Registry クライアント ライブラリをインストールします。

pip install azure-schemaregistry

前提条件:

このパッケージを使用するには、次が必要です。

クライアントを認証する

スキーマ レジストリとの対話は、SchemaRegistryClient クラスのインスタンスから始まります。 クライアント コンストラクターは、完全修飾名前空間と Azure Active Directory 資格情報を受け取ります。

  • スキーマ レジストリ インスタンスの完全修飾名前空間は、 の形式 <yournamespace>.servicebus.windows.netに従う必要があります。

  • TokenCredential プロトコルを実装する AAD 資格情報をコンストラクターに渡す必要があります。 azure-identity パッケージで使用できるプロトコルの実装TokenCredentialがあります。 で azure-identity提供される資格情報の種類を使用するには、 pip を使用して Python 用の Azure Identity クライアント ライブラリをインストールしてください。

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 が提供されます。

例

次のセクションでは、次のような最も一般的なスキーマ レジストリ タスクの一部をカバーするいくつかのコード スニペットを示します。

スキーマを登録する

メソッドを使用して 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 レベルでログに記録されます。

要求/応答本文、未変換ヘッダーなど、詳細な DEBUG レベルのログは、 引数を使用してクライアントで 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 は、詳細なログ記録がクライアントで有効になっていない場合でも、1 回の操作のために有効にすることができます。

schema_registry_client.get_schema(schema_id, logging_enable=True)

次のステップ

その他のサンプル コード

このライブラリを使用してスキーマ レジストリとの間でスキーマを登録および取得する方法の詳細な例については、 サンプル ディレクトリを参照してください。

共同作成

このプロジェクトでは、共同作成と提案を歓迎しています。 ほとんどの共同作成では、共同作成者使用許諾契約書 (CLA) にご同意いただき、ご自身の共同作成内容を使用する権利を Microsoft に供与する権利をお持ちであり、かつ実際に供与することを宣言していただく必要があります。 詳細については、 https://cla.microsoft.com を参照してください。

pull request を送信すると、CLA を提供して PR (ラベル、コメントなど) を適宜装飾する必要があるかどうかを CLA ボットが自動的に決定します。 ボットによって提供される手順にそのまま従ってください。 この操作は、Microsoft の CLA を使用するすべてのリポジトリについて、1 回だけ行う必要があります。

このプロジェクトでは、Microsoft オープン ソースの倫理規定を採用しています。 詳しくは、「Code of Conduct FAQ (倫理規定についてよくある質問)」を参照するか、opencode@microsoft.com 宛てに質問またはコメントをお送りください。