Azure Event Grid klientská knihovna pro Python – verze 4.17.0b1

Článek
11/16/2023

Azure Event Grid je plně spravovaná služba inteligentního směrování událostí, která umožňuje jednotnou spotřebu událostí s využitím modelu publikování a odběru.

Právní omezení

Toto je beta verze služby Azure EventGrid EventGridClient, která spolu s obecně dostupnou EventGridPublisherClientverzí . EventGridClient podporuje publish_cloud_eventsoperace , receive_cloud_events, acknowledge_cloud_events , release_cloud_events, reject_cloud_eventsa renew_cloud_event_locks . Další informace najdete v ukázkách .

Začínáme

Požadavky

K použití tohoto balíčku se vyžaduje Python 3.7 nebo novější.
Abyste mohli tento balíček použít, musíte mít předplatné Azure a prostředek tématu Event Gridu. Postupujte podle tohoto podrobného kurzu a zaregistrujte poskytovatele prostředků Event Gridu a vytvořte témata Event Gridu pomocí Azure Portal. K dispozici je podobný kurz s využitím Azure CLI.

Instalace balíčku

Nainstalujte klientskou knihovnu Azure Event Grid pro Python pomocí pip:

pip install azure-eventgrid

Vyžaduje se existující téma nebo doména Event Gridu. Prostředek můžete vytvořit pomocí webu Azure Portal nebo Azure CLI.

Pokud používáte Azure CLI, nahraďte <resource-group-name> a <resource-name> vlastními jedinečnými názvy.

Vytvoření tématu Event Gridu

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

Vytvoření domény Event Gridu

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

Ověření klienta

Abyste mohli pracovat se službou Event Grid, budete muset vytvořit instanci klienta. K vytvoření instance objektu klienta jsou nezbytné koncové body a přihlašovací údaje .

Použití Azure Active Directory (AAD)

Azure Event Grid poskytuje integraci se službou Azure Active Directory (Azure AD) pro ověřování požadavků na základě identity. S Azure AD můžete pomocí řízení přístupu na základě role (RBAC) udělit uživatelům, skupinám nebo aplikacím přístup k prostředkům Azure Event Grid.

Pokud chcete odesílat události do tématu nebo doméně pomocí TokenCredential, musí mít ověřená identita přiřazenou roli EventGrid Data Sender.

azure-identity Balíček umožňuje bezproblémovou autorizaci požadavků ve vývojovém i produkčním prostředí. Další informace o Azure Active Directory najdete v souboru azure-identity README.

Můžete například použít DefaultAzureCredential k vytvoření klienta, který se bude ověřovat pomocí 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)

Vyhledání koncového bodu

Koncový bod tématu najdete v prostředku Téma Event Gridu na Azure Portal. Bude to vypadat takto: "https://<event-grid-topic-name>.<topic-location>.eventgrid.azure.net/api/events"

Vytvoření klienta pomocí AzureKeyCredential

Pokud chcete jako credential parametr použít přístupový klíč, předejte klíč jako řetězec do instance AzureKeyCredential.

Poznámka: Přístupový klíč najdete na webu Azure Portal v nabídce Přístupové klíče prostředku tématu Event Gridu. Můžete je také získat prostřednictvím azure CLI nebo azure-mgmt-eventgrid knihovny. Průvodce získáním přístupových klíčů najdete tady.

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)

Poznámka: Klient se také může ověřovat prostřednictvím podpisu SAS pomocí AzureSasCredential. Ukázka, která to demonstruje, je k dispozici tady (async_version).

Poznámka: Metodu generate_sas lze použít k vygenerování sdíleného přístupového podpisu. Ukázku, která to demonstruje, najdete tady.

Klíčové koncepty

Téma

Téma je kanál v rámci služby EventGrid pro odesílání událostí. Schéma události, které téma přijímá, se rozhoduje při vytváření tématu. Pokud jsou události typu schématu odeslány do tématu, které vyžaduje jiný typ schématu, budou vyvolány chyby.

Doména

Doména událostí je nástroj pro správu velkého počtu témat Event Gridu souvisejících se stejnou aplikací. Umožňují publikovat události do tisíců témat. Domény také poskytují autorizační a ověřovací kontrolu nad jednotlivými tématy. Další informace najdete v přehledu domény událostí.

Když vytvoříte doménu události, zpřístupní se vám koncový bod publikování pro tuto doménu. Tento proces se podobá vytvoření tématu Event Gridu. Jediným rozdílem je, že při publikování do domény musíte zadat téma v rámci domény, do které chcete událost doručovat.

Schémata událostí

Událost je nejmenší množství informací, které plně popisují něco, co se stalo v systému. Při vytváření vlastního tématu nebo domény musíte zadat schéma, které se použije při publikování událostí.

Event Grid podporuje více schémat pro kódování událostí.

Schéma služby Event Grid

Téma můžete nakonfigurovat tak, aby používalo vlastní schéma, ale běžnější je použít už definované schéma Event Gridu. Specifikace a požadavky najdete tady.

Schéma CloudEvents v1.0

Další možností je použít schéma CloudEvents verze 1.0. CloudEvents je projekt základu nativního výpočetního prostředí pro cloud, který vytváří specifikaci pro popis dat událostí běžným způsobem. Souhrn služby CloudEvents najdete tady.

EventGridPublisherClient

EventGridPublisherClient poskytuje operace pro odesílání dat událostí na název hostitele tématu zadaný během inicializace klienta.

Bez ohledu na schéma, které je pro vaše téma nebo doménu nakonfigurované, EventGridPublisherClient se budou k publikování událostí používat. Použijte metodu send publikování událostí.

Je možné odesílat následující formáty událostí:

Seznam nebo jedna instance EventGridEvents silného typu.
A dict reprezentace serializované EventGridEvent objektu.
Seznam nebo jedna instance CloudEvents silného typu.
dict reprezentace serializované CloudEvent objektu.
Dict reprezentace libovolného vlastního schématu.

Podrobné příklady najdete v ukázkách .

Poznámka: Před publikováním je důležité vědět, jestli vaše téma podporuje CloudEvents nebo EventGridEvents. Pokud zprávu odešlete do tématu, které nepodporuje schéma události, kterou odesíláte, nástroj send() vyvolá výjimku.

Systémová témata

Systémové téma ve službě Event Grid představuje jednu nebo více událostí publikovaných službami Azure, jako je Azure Storage nebo Azure Event Hubs. Systémové téma může například představovat všechny události objektů blob nebo pouze události vytvoření a odstranění objektu blob publikované pro konkrétní účet úložiště.

Názvy různých typů událostí pro systémové události publikované do Azure Event Grid jsou k dispozici v nástroji azure.eventgrid.SystemEventNames. Úplný seznam rozpoznatelných systémových témat najdete v tématu Systémová témata.

Další informace o klíčových konceptech služby Event Grid najdete v tématu Koncepty v Azure Event Grid.

Event Grid v Kubernetes se službou Azure Arc

Event Grid v Kubernetes s Azure Arc je nabídka, která umožňuje spustit Event Grid na vlastním clusteru Kubernetes. Tato funkce je povolená pomocí Kubernetes s podporou Azure Arc. Prostřednictvím Kubernetes s podporou Azure Arc se podporovaný cluster Kubernetes připojuje k Azure. Jakmile se připojíte, budete na něj moct event grid nainstalovat. Další informace najdete tady.

Podpora událostí CNCF v cloudu

Od verze 4.7.0 tento balíček také podporuje publikování cloudové události CNCF z https://pypi.org/project/cloudevents/webu . Z této knihovny send byste mohli rozhraní API předat objekt CloudEvent.


from cloudevents.http import CloudEvent

event = CloudEvent(...)

client.send(event)

Příklady

Následující části obsahují několik fragmentů kódu, které pokrývají některé z nejběžnějších úloh Event Gridu, mezi které patří:

Odeslání události Event Gridu
Odeslání cloudové události
Odeslat více událostí
Odesílání událostí jako slovníků
Využití datové části z fronty úložiště
Využívání ze Služby ServiceBus

Odeslání události Event Gridu

Tento příklad publikuje událost Event Gridu.

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)

Odeslání cloudové události

Tento příklad publikuje událost cloudu.

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)

Odeslání více událostí

Při odesílání více událostí do tématu nebo domény je možné události odesílat jako dávku. Tento příklad odešle seznam CloudEvents pomocí metody send.

UPOZORNĚNÍ: Při odesílání seznamu více událostí najednou nemá iterace a odeslání jednotlivých událostí optimální výkon. Pro zajištění nejlepšího výkonu důrazně doporučujeme odeslat seznam událostí.

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)

Odesílání událostí jako slovníků

Reprezentace dict příslušných serializovaných modelů se dá použít také k publikování CloudEvent(s) nebo EventGridEvent(s) kromě objektů silného typu.

K odeslání do tématu s vlastním schématem použijte reprezentaci jako diktování, jak je znázorněno níže.

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)

Využití z fronty úložiště

Tento příklad využívá zprávu přijatou z fronty úložiště a deserializuje ji do objektu 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]

Využívání ze služby ServiceBus

Tento příklad využívá zprávu datové části přijaté ze služby ServiceBus a deserializuje ji do objektu 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]

Distribuované trasování s využitím EventGridu

OpenTelemetry pro Python můžete s EventGridem používat jako obvykle, protože je kompatibilní s integrací trasování azure-core.

Tady je příklad použití OpenTelemetry k trasování odesílání CloudEvent.

Nejprve nastavte OpenTelemetry jako povolený modul plug-in trasování pro EventGrid.

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

settings.tracing_implementation = OpenTelemetrySpan

Odsud pravidelně používáte otevřenou telemetrii. Podrobnosti najdete v tématu OpenTelemetry . V tomto příkladu se k exportu trasování používá jednoduchý konzolový exportér. Zde může být použit jakýkoli vývozce, včetně azure-monitor-opentelemetry-exporter, jaegeratd zipkin .

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

tracer Po nastavení a exporter postupujte podle následujícího příkladu a začněte shromažďovat trasování při použití send metody z objektu EventGridPublisherClient 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)

Poradce při potížích

Povolte azure.eventgrid protokolovacímu nástroje shromažďovat trasování z knihovny.

Obecné

Klientská knihovna Event Gridu vyvolá výjimky definované v Azure Core.

protokolování

Tato knihovna používá k protokolování standardní knihovnu protokolování . Základní informace o relacích HTTP (adresy URL, hlavičky atd.) se protokolují na úrovni INFO.

Volitelná konfigurace

Volitelné argumenty klíčových slov je možné předat na úrovni klienta a pro jednotlivé operace. Referenční dokumentace azure-core popisuje dostupné konfigurace pro opakování, protokolování, přenosové protokoly a další.

Další kroky

Následující část obsahuje několik fragmentů kódu, které ilustrují běžné vzory používané v rozhraní Python API služby Event Grid.

Další vzorový kód

Tyto ukázky kódu ukazují běžné operace ve scénářích šampionů s klientskou knihovnou Azure Event Grid.

Vygenerovat sdílený přístupový podpis: sample_generate_sas.py
Ověření klienta: sample_authentication.py (async_version)
Publikování událostí do tématu pomocí SAS: sample_publish_events_to_a_topic_using_sas_credential_async.py (async_version)
Publikování událostí Event Gridu do tématu: sample_publish_eg_events_to_a_topic.py (async_version)
Publikování událostí EventGrid do tématu domény: sample_publish_eg_events_to_a_domain_topic.py (async_version)
Publikování cloudové události: sample_publish_events_using_cloud_events_1.0_schema.py (async_version)
Publikování vlastního schématu: sample_publish_custom_schema_to_a_topic.py (async_version)

Následující ukázky popisují publikování a využívání dict reprezentací EventGridEvents a CloudEvents.

Publikování EventGridEvent jako reprezentace dict: sample_publish_eg_event_using_dict.py (async_version)
Publikování CloudEvent jako diktování podobné reprezentaci: sample_publish_cloud_event_using_dict.py (async_version)
Využití vlastní datové části nezpracovaných dat cloudevent: sample_consume_custom_payload.py

Další ukázky najdete tady.

Další ukázky související se scénářem odeslání najdete tady.
Další ukázky související s využíváním datové části z různých služeb zasílání zpráv jako typovaného objektu najdete v tématu Využití ukázek.

Další dokumentace

Podrobnější dokumentaci k Azure Event Grid najdete v dokumentaci ke službě Event Grid k docs.microsoft.com.

Přispívání

Tento projekt vítá příspěvky a návrhy. Většina příspěvků vyžaduje souhlas s licenční smlouvou s přispěvatelem (CLA), která stanoví, že máte právo udělit nám práva k používání vašeho příspěvku a skutečně tak činíte. Podrobnosti najdete v cla.microsoft.com.

Při odesílání žádosti o přijetí změn robot CLA automaticky určí, jestli je potřeba poskytnout smlouvu CLA, a příslušným způsobem žádost o přijetí změn upraví (např. přidáním jmenovky nebo komentáře). Stačí postupovat podle pokynů robota. Pro všechna úložiště používající naši smlouvu CLA to stačí udělat jenom jednou.

Tento projekt přijal pravidla chování pro Microsoft Open Source. Další informace najdete v nejčastějších dotazech k pravidlům chování nebo kontaktujte s opencode@microsoft.com případnými dalšími dotazy nebo připomínkami.

Sdílet prostřednictvím