다음을 통해 공유

Python용 Azure Event Grid 클라이언트 라이브러리 - 버전 4.16.0

  • 아티클

Azure Event Grid는 게시-구독 모델을 사용하여 균일한 이벤트 소비를 허용하는 완전히 관리되는 지능형 이벤트 라우팅 서비스입니다.

소스 코드 | 패키지(PyPI) | 패키지(Conda) | API 참조 설명서 | 제품 설명서 | 샘플 | Changelog

시작

필수 구성 요소

  • 이 패키지를 사용하려면 Python 3.7 이상이 필요합니다.
  • 이 패키지를 사용하려면 Azure 구독 및 Event Grid 토픽 리소스가 있어야 합니다. 이 단계별 자습서에 따라 Event Grid 리소스 공급자를 등록하고 Azure Portal 사용하여 Event Grid topics 만듭니다. Azure CLI를 사용하는 유사한 자습서가 있습니다.

패키지 설치

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

pip install azure-eventgrid
  • 기존 Event Grid 토픽 또는 도메인이 필요합니다. Azure Portal 또는 Azure CLI를 사용하여 리소스를 만들 수 있습니다.

Azure CLI를 사용하는 경우 및 를 <resource-name> 고유한 이름으로 바꿉 <resource-group-name> 니다.

Event Grid 토픽 만들기

az eventgrid topic --create --location <location> --resource-group <resource-group-name> --name <resource-name>

Event Grid 도메인 만들기

az eventgrid domain --create --location <location> --resource-group <resource-group-name> --name <resource-name>

클라이언트 인증

Event Grid 서비스와 상호 작용하려면 클라이언트의 instance 만들어야 합니다. 클라이언트 개체를 인스턴스화하려면 엔드포인트 및 자격 증명 이 필요합니다.

AAD(Azure Active Directory) 사용

Azure Event Grid 요청의 ID 기반 인증을 위해 Azure AD(Azure Active Directory)와 통합됩니다. Azure AD 사용하여 RBAC(역할 기반 액세스 제어)를 사용하여 사용자, 그룹 또는 애플리케이션에 Azure Event Grid 리소스에 대한 액세스 권한을 부여할 수 있습니다.

를 사용하여 토픽 또는 도메인에 TokenCredential이벤트를 보내려면 인증된 ID에 "EventGrid 데이터 보낸 사람" 역할이 할당되어야 합니다.

azure-identity 패키지를 사용하면 개발 환경과 프로덕션 환경 모두에서 요청에 원활하게 권한을 부여할 수 있습니다. Azure Active Directory에 대한 자세한 내용은 추가 정보를azure-identity 참조하세요.

예를 들어 를 사용하여 DefaultAzureCredential Azure Active Directory를 사용하여 인증할 클라이언트를 생성할 수 있습니다.

from azure.identity import DefaultAzureCredential
from azure.eventgrid import EventGridPublisherClient, EventGridEvent

default_az_credential = DefaultAzureCredential()
endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"]
client = EventGridPublisherClient(endpoint, default_az_credential)

엔드포인트 조회

Azure Portal Event Grid 토픽 리소스 내에서 토픽 엔드포인트를 찾을 수 있습니다. 다음과 같습니다. "https://<event-grid-topic-name>.<topic-location>.eventgrid.azure.net/api/events"

AzureKeyCredential을 사용하여 클라이언트 만들기

Access 키를 매개 변수로 credential 사용하려면 AzureKeyCredential의 instance 키를 문자열로 전달합니다.

참고: 액세스 키는 Event Grid 토픽 리소스의 "액세스 키" 메뉴에 있는 Azure Portal에서 찾을 수 있습니다. Azure CLI 또는 azure-mgmt-eventgrid 라이브러리를 통해 가져올 수도 있습니다. 액세스 키를 가져오기 위한 가이드는 여기에서 찾을 수 있습니다.

import os
from azure.eventgrid import EventGridPublisherClient
from azure.core.credentials import AzureKeyCredential

topic_key = os.environ["EVENTGRID_TOPIC_KEY"]
endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"]

credential_key = AzureKeyCredential(topic_key)
client = EventGridPublisherClient(endpoint, credential_key)

참고: 클라이언트는 를 사용하여 AzureSasCredentialSAS 서명을 통해 인증될 수도 있습니다. 이를 보여 주는 샘플은 여기에서 사용할 수 있습니다(async_version).

참고: 메서드를 generate_sas 사용하여 공유 액세스 서명을 생성할 수 있습니다. 이를 보여 주는 샘플은 여기에서 확인할 수 있습니다.

주요 개념

항목

토픽은 이벤트를 보내는 EventGrid 서비스 내의 채널입니다. 토픽이 허용하는 이벤트 스키마는 토픽 생성 시 결정됩니다. 스키마 형식의 이벤트가 다른 스키마 형식이 필요한 토픽으로 전송되면 오류가 발생합니다.

도메인

이벤트 도메인은 동일한 애플리케이션과 관련된 많은 수의 Event Grid topics 대한 관리 도구입니다. 이벤트 도메인을 사용하면 수천 개의 토픽에 이벤트를 게시할 수 있습니다. 또한 도메인은 각 토픽에 대한 권한 부여 및 인증 제어를 제공합니다. 자세한 내용은 이벤트 도메인 개요를 참조하세요.

이벤트 도메인을 만들면 이 도메인에 대한 게시 엔드포인트를 사용할 수 있습니다. 이 프로세스는 Event Grid 토픽을 만드는 것과 비슷합니다. 유일한 차이점은 도메인에 게시할 때 이벤트를 전달하려는 도메인 내에서 토픽을 지정해야 한다는 것입니다.

이벤트 스키마

이벤트는 시스템에서 발생한 일을 완전히 설명하는 가장 적은 양의 정보입니다. 사용자 지정 토픽 또는 도메인을 만들 때 이벤트를 게시할 때 사용할 스키마를 지정해야 합니다.

Event Grid는 이벤트를 인코딩하기 위해 여러 스키마를 지원합니다.

Event Grid 스키마

사용자 지정 스키마를 사용하도록 토픽을 구성할 수 있지만 이미 정의된 Event Grid 스키마를 사용하는 것이 더 일반적입니다. 여기에서 사양 및 요구 사항을 참조 하세요.

CloudEvents v1.0 스키마

또 다른 옵션은 CloudEvents v1.0 스키마를 사용하는 것입니다. CloudEvents 는 일반적인 방식으로 이벤트 데이터를 설명하는 사양을 생성하는 클라우드 네이티브 컴퓨팅 파운데이션 프로젝트입니다. CloudEvents의 서비스 요약은 여기에서 찾을 수 있습니다.

EventGridPublisherClient

EventGridPublisherClient 는 클라이언트 초기화 중에 지정된 토픽 호스트 이름으로 이벤트 데이터를 보내는 작업을 제공합니다.

토픽 또는 도메인이 사용하도록 EventGridPublisherClient 구성된 스키마에 관계없이 이벤트를 게시하는 데 사용됩니다. 이벤트를 게시하는 메서드를 send 사용합니다.

다음과 같은 형식의 이벤트를 보낼 수 있습니다.

  • 강력한 형식의 EventGridEvents의 목록 또는 단일 instance.

  • 직렬화된 EventGridEvent 개체의 받아쓰기 표현입니다.

  • 강력한 형식의 CloudEvents의 목록 또는 단일 instance.

  • 직렬화된 CloudEvent 개체의 받아쓰기 표현입니다.

  • 사용자 지정 스키마의 받아쓰기 표현입니다.

자세한 예제는 샘플을 참조하세요.

참고: 게시하기 전에 토픽이 CloudEvents 또는 EventGridEvents를 지원하는지 아는 것이 중요합니다. 보내는 이벤트의 스키마를 지원하지 않는 토픽으로 보내면 send()에서 예외가 발생합니다.

시스템 토픽

Event Grid의 시스템 항목은 Azure Storage 또는 Azure Event Hubs 같은 Azure 서비스에서 게시한 하나 이상의 이벤트를 나타냅니다. 예를 들어 시스템 토픽은 모든 Blob 이벤트 또는 특정 스토리지 계정에 대해 게시된 Blob 만들기 및 Blob 삭제 이벤트만 나타낼 수 있습니다.

Azure Event Grid 게시된 시스템 이벤트에 대한 다양한 이벤트 유형의 이름은 에서 azure.eventgrid.SystemEventNames사용할 수 있습니다. 인식할 수 있는 시스템 topics 전체 목록은 시스템 항목을 참조하세요.

Event Grid의 주요 개념에 대한 자세한 내용은 Azure Event Grid 개념을 참조하세요.

Azure Arc를 사용하는 Kubernetes의 Event Grid

Azure Arc를 사용하는 Kubernetes의 Event Grid는 자체 Kubernetes 클러스터에서 Event Grid를 실행할 수 있는 제품입니다. 이 기능은 Azure Arc 지원 Kubernetes를 사용하여 활성화됩니다. Azure Arc 지원 Kubernetes를 통해 지원되는 Kubernetes 클러스터가 Azure에 연결됩니다. 연결되면 여기에 Event Grid를 설치할 수 있습니다. 여기에서 자세히 알아보세요.

CNCF 클라우드 이벤트 지원

v4.7.0부터 이 패키지는 에서 https://pypi.org/project/cloudevents/CNCF 클라우드 이벤트 게시도 지원합니다. 이 라이브러리에서 API로 CloudEvent 개체를 send 전달할 수 있습니다.


from cloudevents.http import CloudEvent

event = CloudEvent(...)

client.send(event)

예제

다음 섹션에서는 다음을 포함하여 가장 일반적인 Event Grid 작업을 다루는 몇 가지 코드 조각을 제공합니다.

Event Grid 이벤트 보내기

이 예제에서는 Event Grid 이벤트를 게시합니다.

import os
from azure.core.credentials import AzureKeyCredential
from azure.eventgrid import EventGridPublisherClient, EventGridEvent

key = os.environ["EG_ACCESS_KEY"]
endpoint = os.environ["EG_TOPIC_HOSTNAME"]

event = EventGridEvent(
    data={"team": "azure-sdk"},
    subject="Door1",
    event_type="Azure.Sdk.Demo",
    data_version="2.0"
)

credential = AzureKeyCredential(key)
client = EventGridPublisherClient(endpoint, credential)

client.send(event)

클라우드 이벤트 보내기

이 예제에서는 클라우드 이벤트를 게시합니다.

import os
from azure.core.credentials import AzureKeyCredential
from azure.core.messaging import CloudEvent
from azure.eventgrid import EventGridPublisherClient

key = os.environ["CLOUD_ACCESS_KEY"]
endpoint = os.environ["CLOUD_TOPIC_HOSTNAME"]

event = CloudEvent(
    type="Azure.Sdk.Sample",
    source="https://egsample.dev/sampleevent",
    data={"team": "azure-sdk"}
)

credential = AzureKeyCredential(key)
client = EventGridPublisherClient(endpoint, credential)

client.send(event)

여러 이벤트 보내기

토픽 또는 도메인에 여러 이벤트를 보낼 때 이벤트를 일괄 처리로 보낼 수 있습니다. 이 예제에서는 send 메서드를 사용하여 CloudEvents 목록을 보냅니다.

경고: 한 번에 여러 이벤트 목록을 보내는 경우 반복하고 각 이벤트를 전송해도 성능이 최적화되지 않습니다. 최상의 성능을 위해 이벤트 목록을 보내는 것이 좋습니다.

import os
from azure.core.credentials import AzureKeyCredential
from azure.core.messaging import CloudEvent
from azure.eventgrid import EventGridPublisherClient

key = os.environ["CLOUD_ACCESS_KEY"]
endpoint = os.environ["CLOUD_TOPIC_HOSTNAME"]

event0 = CloudEvent(
    type="Azure.Sdk.Sample",
    source="https://egsample.dev/sampleevent",
    data={"team": "azure-sdk"}
)
event1 = CloudEvent(
    type="Azure.Sdk.Sample",
    source="https://egsample.dev/sampleevent",
    data={"team2": "azure-eventgrid"}
)

events = [event0, event1]

credential = AzureKeyCredential(key)
client = EventGridPublisherClient(endpoint, credential)

client.send(events)

사전으로 이벤트 보내기

직렬화된 각 모델의 받아쓰기 표현을 사용하여 강력한 형식의 개체와 별도로 CloudEvent 또는 EventGridEvent를 게시할 수도 있습니다.

아래와 같이 사용자 지정 스키마를 사용하여 토픽에 보내려면 dict와 유사한 표현을 사용합니다.

import os
import uuid
import datetime as dt
from msrest.serialization import UTC
from azure.core.credentials import AzureKeyCredential
from azure.eventgrid import EventGridPublisherClient

key = os.environ["CUSTOM_SCHEMA_ACCESS_KEY"]
endpoint = os.environ["CUSTOM_SCHEMA_TOPIC_HOSTNAME"]

event = custom_schema_event = {
    "customSubject": "sample",
    "customEventType": "sample.event",
    "customDataVersion": "2.0",
    "customId": uuid.uuid4(),
    "customEventTime": dt.datetime.now(UTC()).isoformat(),
    "customData": "sample data"
    }

credential = AzureKeyCredential(key)
client = EventGridPublisherClient(endpoint, credential)

client.send(event)

스토리지 큐에서 사용

이 예제에서는 스토리지 큐에서 받은 메시지를 사용하고 CloudEvent 개체로 역직렬화합니다.

from azure.core.messaging import CloudEvent
from azure.storage.queue import QueueServiceClient, BinaryBase64DecodePolicy
import os
import json

# all types of CloudEvents below produce same DeserializedEvent
connection_str = os.environ['STORAGE_QUEUE_CONN_STR']
queue_name = os.environ['STORAGE_QUEUE_NAME']

with QueueServiceClient.from_connection_string(connection_str) as qsc:
    payload =  qsc.get_queue_client(
        queue=queue_name,
        message_decode_policy=BinaryBase64DecodePolicy()
        ).peek_messages()

    ## deserialize payload into a list of typed Events
    events = [CloudEvent.from_dict(json.loads(msg.content)) for msg in payload]

servicebus에서 사용

이 예제에서는 ServiceBus에서 받은 페이로드 메시지를 사용하고 EventGridEvent 개체로 역직렬화합니다.

from azure.eventgrid import EventGridEvent
from azure.servicebus import ServiceBusClient
import os
import json

# all types of EventGridEvents below produce same DeserializedEvent
connection_str = os.environ['SERVICE_BUS_CONN_STR']
queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']

with ServiceBusClient.from_connection_string(connection_str) as sb_client:
    payload =  sb_client.get_queue_receiver(queue_name).receive_messages()

    ## deserialize payload into a list of typed Events
    events = [EventGridEvent.from_dict(json.loads(next(msg.body).decode('utf-8'))) for msg in payload]

EventGrid를 사용하여 분산 추적

Azure-core 추적 통합과 호환되므로 EventGrid에서 평소와 같이 Python용 OpenTelemetry를 사용할 수 있습니다.

다음은 OpenTelemetry를 사용하여 CloudEvent 전송을 추적하는 예제입니다.

먼저 OpenTelemetry를 EventGrid에 대해 사용하도록 설정된 추적 플러그 인으로 설정합니다.

from azure.core.settings import settings
from azure.core.tracing.ext.opentelemetry_span import OpenTelemetrySpan

settings.tracing_implementation = OpenTelemetrySpan

여기에서 정기적으로 열린 원격 분석 사용. 자세한 내용은 OpenTelemetry 를 참조하세요. 이 예제에서는 간단한 콘솔 내보내기를 사용하여 추적을 내보냅니다. 여기에서 , , jaegerzipkin 등을 비롯한 azure-monitor-opentelemetry-exporter모든 내보내기를 사용할 수 있습니다.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter
from opentelemetry.sdk.trace.export import SimpleSpanProcessor  # this requires opentelemetry >= 1.0.0

# Simple console exporter
exporter = ConsoleSpanExporter()

trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)
trace.get_tracer_provider().add_span_processor(
    SimpleSpanProcessor(exporter)
)

및 exporter 가 tracer 설정되면 아래 예제에 따라 에서 메서드 EventGridPublisherClient 를 사용하여 send CloudEvent 개체를 보내는 동안 추적 수집을 시작합니다.

import os
from azure.eventgrid import EventGridPublisherClient
from azure.core.messaging import CloudEvent
from azure.core.credentials import AzureKeyCredential

hostname = os.environ['CLOUD_TOPIC_HOSTNAME']
key = AzureKeyCredential(os.environ['CLOUD_ACCESS_KEY'])
cloud_event = CloudEvent(
    source = 'demo',
    type = 'sdk.demo',
    data = {'test': 'hello'},
)
with tracer.start_as_current_span(name="MyApplication"):
    client = EventGridPublisherClient(hostname, key)
    client.send(cloud_event)

문제 해결

  • 로거가 라이브러리에서 추적을 수집하도록 설정합니다 azure.eventgrid .

일반

Event Grid 클라이언트 라이브러리는 Azure Core에 정의된 예외를 발생합니다.

로깅

이 라이브러리는 로깅에 표준 로깅 라이브러리를 사용합니다. HTTP 세션(URL, 헤더 등)에 대한 기본 정보는 INFO 수준에서 기록됩니다.

선택적 구성

선택적 키워드(keyword) 인수는 클라이언트 및 작업별 수준에서 전달할 수 있습니다. azure-core 참조 설명서 에서는 재시도, 로깅, 전송 프로토콜 등에 사용할 수 있는 구성에 대해 설명합니다.

다음 단계

다음 섹션에서는 Event Grid Python API에 사용되는 일반적인 패턴을 보여 주는 몇 가지 코드 조각을 제공합니다.

추가 샘플 코드

이러한 코드 샘플은 Azure Event Grid 클라이언트 라이브러리를 사용하는 일반적인 챔피언 시나리오 작업을 보여 줍니다.

다음 샘플에서는 EventGridEvents 및 CloudEvents의 게시 및 사용 dict 표현을 다룹니다.

추가 샘플은 여기에서 찾을 수 있습니다.

  • 보내기 시나리오와 관련된 더 많은 샘플은 여기에서 확인할 수 있습니다.
  • 다른 메시징 서비스의 페이로드를 형식화된 개체로 사용하는 것과 관련된 더 많은 샘플을 보려면 샘플 사용을 참조하세요.

추가 설명서

Azure Event Grid 대한 보다 광범위한 설명서는 docs.microsoft.com Event Grid 설명서를 참조하세요.

참여

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

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

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