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
, jaeger
atd 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.
Azure SDK for Python
Váš názor
https://aka.ms/ContentUserFeedback.
Připravujeme: V průběhu roku 2024 budeme postupně vyřazovat problémy z GitHub coby mechanismus zpětné vazby pro obsah a nahrazovat ho novým systémem zpětné vazby. Další informace naleznete v tématu:Odeslat a zobrazit názory pro