適用于 Python 的Azure 事件方格用戶端程式庫 - 4.16.0 版

Azure Event Grid 是完全受控的智慧型事件路由服務，可讓統一事件耗用量使用發佈-訂閱模型。

| 原始程式碼套件 (PyPI) | 套件 (Conda) | API 參考檔 | 產品檔 | 樣品 | Changelog

開始使用

Prerequisites

• 需要 Python 3.7 或更新版本才能使用此套件。
• 您必須擁有 Azure 訂 用帳戶和事件方格主題資源，才能使用此套件。 請遵循本逐步教學課程，以註冊事件方格資源提供者，並使用Azure 入口網站建立事件方格主題。 使用Azure CLI有類似的教學課程。

安裝套件

使用pip安裝適用于 Python 的Azure 事件方格用戶端程式庫：

pip install azure-eventgrid

• 需要現有的事件方格主題或網域。 您可以使用Azure 入口網站或Azure CLI建立資源

如果您使用 Azure CLI，請將 和 <resource-name> 取代 <resource-group-name> 為您自己的唯一名稱。

建立事件方格主題

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

建立事件方格網域

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

驗證用戶端

若要與事件方格服務互動，您必須建立用戶端的實例。 必須有 端點 和 認證 ，才能具現化用戶端物件。

使用 Azure Active Directory (AAD)

Azure 事件方格提供與 Azure Active Directory (Azure AD) 的整合，以進行要求的身分識別型驗證。 透過 Azure AD，您可以使用角色型存取控制 (RBAC) ，將Azure 事件方格資源的存取權授與使用者、群組或應用程式。

若要將事件傳送至具有 TokenCredential 的主題或網域，已驗證的身分識別應該已指派「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 入口網站的主題端點。 這看起來會像這樣： "https://<event-grid-topic-name>.<topic-location>.eventgrid.azure.net/api/events"

使用 AzureKeyCredential 建立用戶端

若要使用 Access 金鑰做為參數，請將金鑰當做 credential 字串傳遞至 AzureKeyCredential的實例。

注意： 存取金鑰可以在 Azure 入口網站的 [事件方格主題] 資源的 [存取金鑰] 功能表中找到。 您也可以透過 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)

注意： 用戶端也可以使用 透過 SAS 簽章 AzureSasCredential 進行驗證。 示範此範例的範例可 在這裡 取得 (async_version) 。

注意： 方法 generate_sas 可用來產生共用存取簽章。 您可以在這裡看到示範這一點的範例。

重要概念

主題

主題是 EventGrid 服務內用來傳送事件的通道。 主題接受的事件架構會在建立主題時決定。 如果架構類型的事件傳送至需要不同架構類型的主題，將會引發錯誤。

網域

事件 網域 是一種管理工具，適用于與相同應用程式相關的大量事件方格主題。 它們讓您能夠將事件發行到數千個主題。 網域也可讓您對每個主題進行授權和驗證控制。 如需詳細資訊，請流覽 事件網域概觀。

當您建立事件網域時，此網域的發佈端點可供您使用。 此程式類似于建立事件方格主題。 唯一的差別在於，發佈至網域時，您必須在您想要傳遞事件的網域內指定主題。

事件結構描述

事件是最少量的資訊，可完整描述系統中發生的情況。 建立自訂主題或網域時，您必須指定發行事件時將使用的架構。

事件方格支援多個編碼事件的架構。

事件格線結構描述

雖然您可以將主題設定為使用 自訂架構，但使用已定義的事件方格架構比較常見。 請參閱 這裡的規格和需求。

CloudEvents v1.0 結構描述

另一個選項是使用 CloudEvents v1.0 架構。 CloudEvents 是 Cloud Native Computing Foundation 專案，其會產生一個規格，以常見方式描述事件資料。 您可以在這裡找到 CloudEvents 的服務摘要。

EventGridPublisherClient

EventGridPublisherClient 提供作業，以將事件資料傳送至用戶端初始化期間所指定的主題主機名稱。

無論您的主題或網域設定為使用何種架構， EventGridPublisherClient 都會用來將事件發佈至該架構。 send使用 方法發佈事件。

允許傳送下列格式的事件：

• 強型別 EventGridEvents 的清單或單一實例。

• 序列化 EventGridEvent 物件的聽寫表示。

• 強型別 CloudEvents 的清單或單一實例。

• 序列化 CloudEvent 物件的聽寫表示。

• 任何自訂架構的聽寫表示。

請查看 範例 以取得詳細範例。

注意： 請務必知道您的主題是否在發佈之前支援 CloudEvents 或 EventGridEvents。 如果您傳送至不支援所傳送事件架構的主題，send () 將會擲回例外狀況。

系統主題

事件方格中的系統主題代表 Azure 服務所發佈的一或多個事件，例如 Azure 儲存體或Azure 事件中樞。 例如，系統主題可能代表所有 Blob 事件，或只代表針對特定儲存體帳戶發佈的 Blob 建立和 Blob 刪除事件。

中提供 azure.eventgrid.SystemEventNames 發行至 Azure 事件方格之系統事件的各種事件種類名稱。 如需可辨識系統主題的完整清單，請流覽 系統主題。

如需事件方格上重要概念的詳細資訊，請參閱Azure 事件方格中的概念。

具有 Azure Arc 的 Kubernetes 上的事件方格

具有 Azure Arc 的 Kubernetes 上的事件方格是一個供應項目，可讓您在自己的 Kubernetes 叢集上執行事件方格。 此功能是透過使用已啟用 Azure Arc 的 Kubernetes 來啟用。 透過已啟用 Azure Arc 的 Kubernetes，支援的 Kubernetes 叢集會連線到 Azure。 連線之後，您就可以在其上安裝事件方格。 在此深入了解。

支援針對#2012 年 12 月的雲端活動

從 4.7.0 版開始，此套件也支援從 https://pypi.org/project/cloudevents/ 發佈一個特定版的一個特定活動。 您可以將 CloudEvent 物件從此程式庫傳遞至 send API。

from cloudevents.http import CloudEvent

event = CloudEvent(...)

client.send(event)

範例

下列各節提供數個程式碼片段，涵蓋一些最常見的事件方格工作，包括：

• 傳送事件方格事件
• 傳送雲端事件
• 傳送多個事件
• 將事件當做字典傳送
• 從儲存體佇列取用承載
• 從 ServiceBus 取用

傳送事件方格事件

此範例會發佈 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)

傳送雲端事件

此範例會發佈 Cloud 事件。

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 (s) 或 EventGridEvent (s) 與強型別物件不同。

使用類似聽寫的標記法傳送至具有自訂架構的主題，如下所示。

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 進行分散式追蹤

您可以如往常使用適用于 Python 的 OpenTelemetry 搭配 EventGrid，因為它與 azure 核心追蹤整合相容。

以下是使用 OpenTelemetry 追蹤傳送 CloudEvent 的範例。

首先，將 OpenTelemetry 設定為 EventGrid 的已啟用追蹤外掛程式。

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

settings.tracing_implementation = OpenTelemetrySpan

從這裡定期開啟遙測使用量。 如需詳細資訊，請參閱 OpenTelemetry 。 此範例使用簡單的主控台匯出工具匯出追蹤。 這裡可以使用任何匯出工具，包括 azure-monitor-opentelemetry-exporter 、 jaegerzipkin 等等。

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設定 和 exporter 之後，請遵循下列範例，在從 中使用 send 方法來 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)

疑難排解

• 啟用 azure.eventgrid 記錄器從程式庫收集追蹤。

一般

事件方格用戶端程式庫將會引發 Azure Core中定義的例外狀況。

記錄

此程式庫會使用標準 記錄 程式庫進行記錄。 HTTP 會話的基本資訊 (URL、標頭等 ) 會記錄在 INFO 層級。

選用組態

選擇性關鍵字引數可以在用戶端和每個作業層級傳入。 azure 核心 參考檔 說明重試、記錄、傳輸通訊協定等可用的組態。

下一步

下一節提供數個程式碼片段，說明事件方格 Python API 中使用的常見模式。

更多的程式碼範例

這些程式碼範例示範使用 Azure 事件方格 用戶端程式庫的常見風雲人物案例作業。

• 產生共用存取簽章： sample_generate_sas.py

• 驗證用戶端：sample_authentication.py (async_version)

• 使用 SAS 將事件發佈至主題：sample_publish_events_to_a_topic_using_sas_credential_async.py (async_version)

• 將事件方格事件發佈至主題：sample_publish_eg_events_to_a_topic.py (async_version)

• 將 EventGrid 事件發佈至網域主題：sample_publish_eg_events_to_a_domain_topic.py (async_version)

• 發佈雲端事件：sample_publish_events_using_cloud_events_1.0_schema.py (async_version)

• 發佈自訂架構：sample_publish_custom_schema_to_a_topic.py (async_version)

下列範例涵蓋發行及取用 dict EventGridEvents 和 CloudEvents 的標記法。

• 將 EventGridEvent 發佈為如標記法的聽寫： sample_publish_eg_event_using_dict.py (async_version)

• 將 CloudEvent 發佈為如標記法的聽寫：sample_publish_cloud_event_using_dict.py (async_version)

• 取用原始 cloudevent 資料的自訂承載： sample_consume_custom_payload.py

您可以 在這裡找到更多範例。

• 您可以在這裡看到更多與傳送案例相關的範例。
• 若要查看從不同傳訊服務取用承載作為具型別物件的相關更多範例，請流覽 取用範例

其他文件

如需Azure 事件方格的更廣泛檔，請參閱 docs.microsoft.com上的事件方格檔。

參與

此專案歡迎參與和提供建議。 大部分的參與都要求您同意「參與者授權合約 (CLA)」，宣告您有權且確實授與我們使用投稿的權利。 如需詳細資訊，請造訪 cla.microsoft.com。

當您提交提取要求時，CLA Bot 會自動判斷您是否需要提供 CLA，並適當地裝飾 PR (例如標籤、註解)。 請遵循 bot 提供的指示。 您只需要使用我們的 CLA 在所有存放庫上執行此動作一次。

此專案採用 Microsoft Open Source Code of Conduct (Microsoft 開放原始碼管理辦法)。 如需詳細資訊，請參閱管理辦法常見問題集，如有任何其他問題或意見請連絡 opencode@microsoft.com。