Udostępnij za pośrednictwem

biblioteka klienta Azure Event Grid dla języka Python — wersja 4.16.0

  • Artykuł

Usługa Azure Event Grid to w pełni zarządzana usługa inteligentnego routingu zdarzeń, która umożliwia ujednolicone używanie zdarzeń za pomocą modelu opartego na publikowaniu i subskrybowaniu.

Kod | źródłowyPakiet (PyPI) | Pakiet (Conda) | Dokumentacja referencyjna interfejsu | APIDokumentacja | produktuPróbki | Changelog

Wprowadzenie

Wymagania wstępne

Instalowanie pakietu

Zainstaluj bibliotekę klienta Azure Event Grid dla języka Python za pomocą narzędzia pip:

pip install azure-eventgrid

Jeśli używasz interfejsu wiersza polecenia platformy Azure, zastąp <resource-group-name> wartości i <resource-name> własnymi unikatowymi nazwami.

Tworzenie tematu usługi Event Grid

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

Tworzenie domeny usługi Event Grid

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

Uwierzytelnianie klienta

Aby wchodzić w interakcje z usługą Event Grid, należy utworzyć wystąpienie klienta. Punkt końcowy i poświadczenia są niezbędne do utworzenia wystąpienia obiektu klienta.

Korzystanie z usługi Azure Active Directory (AAD)

Azure Event Grid zapewnia integrację z usługą Azure Active Directory (Azure AD) na potrzeby uwierzytelniania opartego na tożsamościach żądań. Za pomocą Azure AD można użyć kontroli dostępu opartej na rolach (RBAC) w celu udzielenia dostępu do zasobów Azure Event Grid użytkownikom, grupom lub aplikacjom.

Aby wysyłać zdarzenia do tematu lub domeny za pomocą TokenCredentialelementu , tożsamość uwierzytelniona powinna mieć przypisaną rolę "Nadawca danych usługi EventGrid".

azure-identity Pakiet umożliwia bezproblemowe autoryzowanie żądań zarówno w środowiskach deweloperskich, jak i produkcyjnych. Aby dowiedzieć się więcej o usłudze Azure Active Directory, zobacz azure-identity PLIK README.

Na przykład można użyć DefaultAzureCredential polecenia , aby skonstruować klienta, który będzie uwierzytelniany przy użyciu usługi 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)

Szukanie punktu końcowego

Punkt końcowy tematu można znaleźć w zasobie tematu usługi Event Grid w Azure Portal. Będzie to wyglądać następująco: "https://<event-grid-topic-name>.<topic-location>.eventgrid.azure.net/api/events"

Tworzenie klienta za pomocą elementu AzureKeyCredential

Aby użyć klucza dostępu jako parametru credential , przekaż klucz jako ciąg do wystąpienia obiektu AzureKeyCredential.

Uwaga: Klucz dostępu można znaleźć w witrynie Azure Portal w menu "Klucze dostępu" zasobu tematu usługi Event Grid. Można je również uzyskać za pośrednictwem interfejsu wiersza polecenia platformy Azure lub biblioteki azure-mgmt-eventgrid . Przewodnik dotyczący pobierania kluczy dostępu można znaleźć tutaj.

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)

Uwaga: Klient może być również uwierzytelniany za pomocą sygnatury sygnatury dostępu współdzielonego przy użyciu polecenia AzureSasCredential. Przykład pokazujący to jest dostępny tutaj (async_version).

Uwaga: Metoda generate_sas może służyć do generowania sygnatury dostępu współdzielonego. Przykład pokazujący to można zobaczyć tutaj.

Kluczowe pojęcia

Temat

Temat jest kanałem w usłudze EventGrid do wysyłania zdarzeń. Schemat zdarzeń akceptowany przez temat jest ustalany w czasie tworzenia tematu. Jeśli zdarzenia typu schematu są wysyłane do tematu, który wymaga innego typu schematu, zostaną zgłoszone błędy.

Domena

Domena zdarzeń to narzędzie do zarządzania dla dużej liczby tematów usługi Event Grid związanych z tą samą aplikacją. Umożliwiają one publikowanie zdarzeń w tysiącach tematów. Domeny zapewniają również autoryzację i kontrolę uwierzytelniania nad każdym tematem. Aby uzyskać więcej informacji, odwiedź stronę Omówienie domeny zdarzeń.

Podczas tworzenia domeny zdarzeń zostanie udostępniony punkt końcowy publikowania dla tej domeny. Ten proces jest podobny do tworzenia tematu usługi Event Grid. Jedyną różnicą jest to, że podczas publikowania w domenie należy określić temat w domenie, do której ma zostać dostarczone zdarzenie.

Schematy zdarzeń

Zdarzenie to najmniejsza ilość informacji, które w pełni opisują coś, co wydarzyło się w systemie. Podczas tworzenia tematu niestandardowego lub domeny należy określić schemat, który będzie używany podczas publikowania zdarzeń.

Usługa Event Grid obsługuje wiele schematów dla zdarzeń kodowania.

Schemat usługi Event Grid

Chociaż temat można skonfigurować tak, aby używał schematu niestandardowego, częściej używa się już zdefiniowanego schematu usługi Event Grid. Zobacz specyfikacje i wymagania tutaj.

Schemat CloudEvents w wersji 1.0

Inną opcją jest użycie schematu CloudEvents w wersji 1.0. CloudEvents to projekt Cloud Native Computing Foundation, który tworzy specyfikację do opisywania danych zdarzeń w typowy sposób. Podsumowanie usługi CloudEvents można znaleźć tutaj.

EventGridPublisherClient

EventGridPublisherClient Udostępnia operacje wysyłania danych zdarzeń do nazwy hosta tematu określonej podczas inicjowania klienta.

Niezależnie od schematu, który jest skonfigurowany do użycia w temacie lub domenie, EventGridPublisherClient będzie używany do publikowania zdarzeń. Użyj metody publikowania send zdarzeń.

Można wysyłać następujące formaty zdarzeń:

  • Lista lub pojedyncze wystąpienie silnie typizowanego zdarzenia EventGridEvents.

  • Reprezentacja dyktowanego obiektu EventGridEvent serializowanego.

  • Lista lub pojedyncze wystąpienie silnie typizowanego rozwiązania CloudEvents.

  • Reprezentacja dyktowanego obiektu CloudEvent serializowanego.

  • Dyktowanie reprezentacji dowolnego schematu niestandardowego.

Zapoznaj się z przykładami , aby zapoznać się ze szczegółowymi przykładami.

Uwaga: Przed opublikowaniem należy wiedzieć, czy temat obsługuje usługę CloudEvents lub EventGridEvents. Jeśli wyślesz do tematu, który nie obsługuje schematu wysyłanego zdarzenia, funkcja send() zgłosi wyjątek.

Tematy systemowe

Temat systemowy w usłudze Event Grid reprezentuje co najmniej jedno zdarzenie opublikowane przez usługi platformy Azure, takie jak Azure Storage lub Azure Event Hubs. Na przykład temat systemowy może reprezentować wszystkie zdarzenia obiektów blob lub tylko zdarzenia tworzenia obiektów blob i usuwania obiektów blob opublikowane dla określonego konta magazynu.

Nazwy różnych typów zdarzeń dla zdarzeń systemowych publikowanych w Azure Event Grid są dostępne w programie azure.eventgrid.SystemEventNames. Aby uzyskać pełną listę rozpoznawalnych tematów systemowych, odwiedź stronę Tematy systemowe.

Aby uzyskać więcej informacji na temat kluczowych pojęć w usłudze Event Grid, zobacz Pojęcia w Azure Event Grid.

Usługa Event Grid na platformie Kubernetes z usługą Azure Arc

Usługa Event Grid na platformie Kubernetes z usługą Azure Arc to oferta umożliwiająca uruchamianie usługi Event Grid we własnym klastrze Kubernetes. Ta funkcja jest włączana przy użyciu platformy Kubernetes z włączoną usługą Azure Arc. Za pośrednictwem platformy Kubernetes z włączoną usługą Azure Arc obsługiwany klaster Kubernetes łączy się z platformą Azure. Po nawiązaniu połączenia możesz zainstalować usługę Event Grid. Więcej informacji na ten temat znajduje się tutaj.

Obsługa zdarzeń w chmurze PLATFORMy CNCF

Począwszy od wersji 4.7.0, ten pakiet obsługuje również publikowanie zdarzenia chmury CNCF z witryny https://pypi.org/project/cloudevents/. Możesz przekazać obiekt CloudEvent z tej biblioteki do interfejsu send API.


from cloudevents.http import CloudEvent

event = CloudEvent(...)

client.send(event)

Przykłady

W poniższych sekcjach przedstawiono kilka fragmentów kodu obejmujących niektóre z najbardziej typowych zadań usługi Event Grid, w tym:

Wysyłanie zdarzenia usługi Event Grid

Ten przykład publikuje zdarzenie usługi 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)

Wysyłanie zdarzenia w chmurze

W tym przykładzie publikowane jest zdarzenie w chmurze.

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)

Wysyłanie wielu zdarzeń

Istnieje możliwość wysyłania zdarzeń jako partii podczas wysyłania wielu zdarzeń do tematu lub domeny. W tym przykładzie jest wysyłana lista rozwiązań CloudEvents przy użyciu metody send.

OSTRZEŻENIE: Podczas wysyłania listy wielu zdarzeń jednocześnie iteracja i wysyłanie każdego zdarzenia nie spowoduje optymalnej wydajności. Aby uzyskać najlepszą wydajność, zdecydowanie zaleca się wysłanie listy zdarzeń.

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)

Wysyłanie zdarzeń jako słowników

Dyktowa reprezentacja odpowiednich serializowanych modeli może być również używana do publikowania obiektów CloudEvent lub EventGridEvent oprócz silnie typizowanych obiektów.

Użyj reprezentacji przypominającej dykt, aby wysłać do tematu z niestandardowym schematem, jak pokazano poniżej.

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)

Korzystanie z kolejki magazynu

W tym przykładzie jest używany komunikat odebrany z kolejki magazynu i deserializuje go do obiektu 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]

Korzystanie z usługi ServiceBus

W tym przykładzie jest używany komunikat ładunku odebrany z usługi ServiceBus i deserializuje go do obiektu 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]

Śledzenie rozproszone za pomocą usługi EventGrid

Możesz użyć biblioteki OpenTelemetry dla języka Python jak zwykle z usługą EventGrid, ponieważ jest ona zgodna z integracją śledzenia podstawowego platformy Azure.

Oto przykład użycia biblioteki OpenTelemetry do śledzenia wysyłania rozwiązania CloudEvent.

Najpierw ustaw wartość OpenTelemetry jako włączoną wtyczkę śledzenia dla usługi EventGrid.

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

settings.tracing_implementation = OpenTelemetrySpan

Regularne otwieranie użycia telemetrii z tego miejsca. Aby uzyskać szczegółowe informacje, zobacz OpenTelemetry . W tym przykładzie użyto prostego eksportera konsoli do wyeksportowania śladów. W tym miejscu można użyć dowolnego eksportera, w tym azure-monitor-opentelemetry-exporter, jaegerzipkin itp.

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)
)

Po ustawieniu tracer elementu i exporter postępuj zgodnie z poniższym przykładem, aby rozpocząć zbieranie śladów przy użyciu send metody z EventGridPublisherClient metody , aby wysłać obiekt 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)

Rozwiązywanie problemów

  • Włącz azure.eventgrid rejestrator w celu zbierania śladów z biblioteki.

Ogólne

Biblioteka klienta usługi Event Grid zgłosi wyjątki zdefiniowane w usłudze Azure Core.

Rejestrowanie

Ta biblioteka używa standardowej biblioteki rejestrowania do rejestrowania. Podstawowe informacje o sesjach HTTP (adresach URL, nagłówkach itp.) są rejestrowane na poziomie INFORMACJI.

Konfiguracja opcjonalna

Opcjonalne argumenty słów kluczowych można przekazać na poziomie klienta i na poziomie operacji. W dokumentacji referencyjnej platformy Azure opisano dostępne konfiguracje ponownych prób, rejestrowania, protokołów transportowych i nie tylko.

Następne kroki

W poniższej sekcji przedstawiono kilka fragmentów kodu ilustrujących typowe wzorce używane w interfejsie API języka Python usługi Event Grid.

Więcej przykładów kodu

Te przykłady kodu pokazują typowe operacje scenariusza championa w bibliotece klienta Azure Event Grid.

Poniższe przykłady obejmują publikowanie i używanie dict reprezentacji usługi EventGridEvents i CloudEvents.

Więcej przykładów można znaleźć tutaj.

  • Więcej przykładów związanych ze scenariuszem wysyłania można znaleźć tutaj.
  • Aby wyświetlić więcej przykładów związanych z używaniem ładunku z różnych usług obsługi komunikatów jako obiektu wpisanego, odwiedź stronę Korzystanie z przykładów

Dodatkowa dokumentacja

Aby uzyskać bardziej obszerną dokumentację dotyczącą Azure Event Grid, zobacz dokumentację usługi Event Grid dotyczącą docs.microsoft.com.

Współtworzenie

W tym projekcie zachęcamy do współtworzenia i zgłaszania sugestii. Współtworzenie w większości przypadków wymaga zgody na umowę licencyjną dotyczącą współautorów (CLA, Contributor License Agreement), zgodnie z którą współautor ma prawo udzielić i faktycznie udziela nam praw do używania wytworzonej przez siebie zawartości. Aby uzyskać szczegółowe informacje, odwiedź stronę cla.microsoft.com.

Po przesłaniu żądania ściągnięcia robot CLA automatycznie określi, czy musisz przekazać umowę CLA, i doda odpowiednie informacje do tego żądania (na przykład etykietę czy komentarz). Po prostu postępuj zgodnie z instrukcjami robota. Wystarczy zrobić to raz dla wszystkich repozytoriów, w przypadku których jest używana nasza umowa CLA.

W tym projekcie przyjęto Kodeks postępowania oprogramowania Open Source firmy Microsoft. Aby uzyskać więcej informacji, zobacz Często zadawane pytania dotyczące kodeksu postępowania lub skontaktuj się z opencode@microsoft.com dodatkowymi pytaniami lub komentarzami.