Python용 Azure Schema Registry 클라이언트 라이브러리 - 버전 1.2.0

아티클
04/05/2023

Azure Schema Registry는 Azure Event Hubs 호스팅하는 스키마 리포지토리 서비스로, 스키마 스토리지, 버전 관리 및 관리를 제공합니다. 레지스트리는 전체 스키마가 아닌 스키마 식별자를 사용하여 페이로드 구조를 설명하는 동안 직렬 변환기에서 페이로드 크기를 줄이는 데 활용됩니다.

소스 코드 | 패키지(PyPi) | API 참조 설명서 | 샘플 | Changelog

고지 사항

Python 2.7에 대한 Azure SDK Python 패키지 지원은 2022년 1월 1일 01일에 종료되었습니다. 자세한 내용과 질문은 다음을 참조하세요. https://github.com/Azure/azure-sdk-for-python/issues/20691

시작

패키지 설치

pip를 사용하여 Python용 Azure Schema Registry 클라이언트 라이브러리를 설치합니다.

pip install azure-schemaregistry

필수 조건:

이 패키지를 사용하려면 다음이 있어야 합니다.

Azure 구독 -체험 계정 만들기
Azure 스키마 레지스트리 - 다음은 Azure Portal 사용하여 스키마 레지스트리 그룹을 만드는 빠른 시작 가이드입니다.
Python 3.7 이상 - Python 설치

클라이언트 인증

스키마 레지스트리와의 상호 작용은 SchemaRegistryClient 클래스의 인스턴스로 시작됩니다. 클라이언트 생성자는 정규화된 네임스페이스와 Azure Active Directory 자격 증명을 사용합니다.

스키마 레지스트리 인스턴스의 정규화된 네임스페이스는 형식 <yournamespace>.servicebus.windows.net을 따라야 합니다.
TokenCredential 프로토콜을 구현하는 AAD 자격 증명을 생성자에 전달해야 합니다. azure-identity 패키지에서 사용할 수 있는 프로토콜의 TokenCredential 구현이 있습니다. 가 제공하는 azure-identity자격 증명 형식을 사용하려면 pip를 사용하여 Python용 Azure ID 클라이언트 라이브러리를 설치하세요.

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 수준에서 기록됩니다.

요청/응답 본문 및 수정되지 않은 헤더를 포함한 자세한 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은 클라이언트에 대해 상세 로깅을 사용하지 않는 경우에도 한 작업에만 사용하게 설정할 수 있습니다.

schema_registry_client.get_schema(schema_id, logging_enable=True)

다음 단계

추가 샘플 코드

이 라이브러리를 사용하여 스키마 레지스트리에 스키마를 등록하고 검색하는 방법에 대한 자세한 예제는 샘플 디렉터리를 참조하세요.

참여

이 프로젝트에 대한 기여와 제안을 환영합니다. 대부분의 경우 기여하려면 권한을 부여하며 실제로 기여를 사용할 권한을 당사에 부여한다고 선언하는 CLA(기여자 라이선스 계약)에 동의해야 합니다. 자세한 내용은 https://cla.microsoft.com 을 참조하세요.

끌어오기 요청을 제출하면 CLA-bot은 CLA를 제공하고 PR을 적절하게 데코레이팅해야 하는지 여부를 자동으로 결정합니다(예: 레이블, 설명). 봇에서 제공하는 지침을 따르기만 하면 됩니다. 이 작업은 CLA를 사용하여 모든 리포지토리에서 한 번만 수행하면 됩니다.

이 프로젝트에는 Microsoft Open Source Code of Conduct(Microsoft 오픈 소스 준수 사항)가 적용됩니다. 자세한 내용은 Code of Conduct FAQ(규정 FAQ)를 참조하세요. 또는 추가 질문이나 의견은 opencode@microsoft.com으로 문의하세요.

다음을 통해 공유