Azure Event Grid klientské knihovny pro Python – verze 4.16.0

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.

Zdrojový kód | Balíček (PyPI) | Balíček (Conda) | Referenční dokumentace k | rozhraní APIDokumentace k | produktuVzorky | Changelog

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žívat, musíte mít předplatné Azure a prostředek tématu Event Gridu. Podle tohoto podrobného kurzu zaregistrujte poskytovatele prostředků Event Gridu a pomocí Azure Portal vytvořte témata event gridu. K dispozici je podobný kurz s využitím Azure CLI.

Instalace balíčku

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

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 Služby Event Grid

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. Koncový bod a přihlašovací údaje jsou nezbytné k vytvoření instance objektu klienta.

Použití Azure Active Directory (AAD)

Azure Event Grid poskytuje integraci s 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ény pomocí TokenCredential, musí mít ověřená identita přiřazenou roli EventGrid Data Sender.

S balíčkem azure-identity můžete bez problémů autorizovat požadavky ve vývojovém i produkčním prostředí. Další informace o Službě Azure Active Directory najdete v azure-identity tématu 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 rámci prostředku Event Grid Topic na Azure Portal. Bude 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 může být také ověřen 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í. O schématu událostí, které téma přijímá, se rozhodne 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 pro tisíce témat. Domény také poskytují autorizaci a ověřování řízení nad jednotlivými tématy. Další informace najdete v tématu Přehled domény událostí.

Když vytvoříte doménu událostí, 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 doméně, 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 je nutné 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 sice můžete nakonfigurovat tak, aby používalo vlastní schéma, ale běžnější je použít již definované schéma Event Gridu. Specifikace a požadavky najdete tady.

Schéma CloudEvents verze 1.0

Další možností je použít schéma CloudEvents verze 1.0. CloudEvents je projekt cloudových nativních výpočetních prostředků, který vytváří specifikaci pro běžné popisování dat událostí. Souhrn služby CloudEvents najdete tady.

EventGridPublisherClient

EventGridPublisherClient poskytuje operace pro odesílání dat událostí do názvu hostitele tématu zadaného během inicializace klienta.

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

Můžou se odesílat následující formáty událostí:

• Seznam nebo jedna instance silného typu EventGridEvents.

• A dict reprezentace serializované EventGridEvent objektu.

• Seznam nebo jedna instance CloudEvents se silnými typy.

• Reprezentace diktu serializovaného objektu CloudEvent.

• Reprezentace diktu 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 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í objektů 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 ve službě Event Grid najdete v tématu Koncepty v Azure Event Grid.

Event Grid v Kubernetes s využitím Azure Arc

Event Grid v Kubernetes s Azure Arc je nabídka, která umožňuje spouštět 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 k Azure připojuje podporovaný cluster Kubernetes. Po připojení na něj můžete nainstalovat Event Grid. 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/. 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é nejběžnější úlohy 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í ve slovníkech
• 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 cloudovou 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"]

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)

Odeslat více událostí

Při odesílání více událostí do tématu nebo domény je možné odesílat události 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 nebude iterace a odeslání každé události mít za následek 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í ve slovníkech

K publikování CloudEvent(s) nebo EventGridEvent (objektů) se silnými typy lze použít také reprezentaci diktování příslušných serializovaných modelů.

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 servicebusu

Tento příklad využívá datovou část zprávu přijatou z 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

S EventGridem můžete jako obvykle používat OpenTelemetry pro Python, 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ý export konzoly. 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ástroji shromažďování trasování z knihovny.

Obecné

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

protokolování

Tato knihovna používá pro 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 na úrovni jednotlivých operací. 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í Api Pythonu služby Event Grid.

Další ukázkový kód

Tyto ukázky kódu ukazují běžné operace scénářů š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 se zabývají publikováním a využíváním dict reprezentací EventGridEvents a CloudEvents.

• Publikování EventGridEvent jako reprezentace diktu: sample_publish_eg_event_using_dict.py (async_version)

• Publikování CloudEventu jako diktování jako reprezentace: sample_publish_cloud_event_using_dict.py (async_version)

• Využití vlastní datové části nezpracovaných dat cloudeventu: sample_consume_custom_payload.py

Další ukázky najdete tady.

• Další ukázky související se scénářem odeslání najdete tady.
• Pokud chcete zobrazit 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, navštivte téma Využití ukázek.

Další dokumentace

Rozsáhlejší 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 na 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 se obraťte na opencode@microsoft.com případné další dotazy nebo komentáře.