azure-eventgrid-py

Azure Event Grid SDK for Python

pub/sub セマンティクスでイベント駆動型アプリケーションを構築するためのイベントルーティングサービスです。

インストール

pip install azure-eventgrid azure-identity

環境変数

EVENTGRID_TOPIC_ENDPOINT=https://<topic-name>.<region>.eventgrid.azure.net/api/events
EVENTGRID_NAMESPACE_ENDPOINT=https://<namespace>.<region>.eventgrid.azure.net

認証

from azure.identity import DefaultAzureCredential
from azure.eventgrid import EventGridPublisherClient

credential = DefaultAzureCredential()
endpoint = "https://<topic-name>.<region>.eventgrid.azure.net/api/events"

client = EventGridPublisherClient(endpoint, credential)

イベントの種類

形式	クラス	用途
Cloud Events 1.0	CloudEvent	標準的で相互運用性に優れています（推奨）
Event Grid スキーマ	EventGridEvent	Azure ネイティブ形式

CloudEvents の発行

from azure.eventgrid import EventGridPublisherClient
from azure.core.messaging import CloudEvent
from azure.identity import DefaultAzureCredential

client = EventGridPublisherClient(endpoint, DefaultAzureCredential())

# Single event
event = CloudEvent(
    type="MyApp.Events.OrderCreated",
    source="/myapp/orders",
    data={"order_id": "12345", "amount": 99.99}
)
client.send(event)

# Multiple events
events = [
    CloudEvent(
        type="MyApp.Events.OrderCreated",
        source="/myapp/orders",
        data={"order_id": f"order-{i}"}
    )
    for i in range(10)
]
client.send(events)

EventGridEvents の発行

from azure.eventgrid import EventGridEvent
from datetime import datetime, timezone

event = EventGridEvent(
    subject="/myapp/orders/12345",
    event_type="MyApp.Events.OrderCreated",
    data={"order_id": "12345", "amount": 99.99},
    data_version="1.0"
)

client.send(event)

イベントのプロパティ

CloudEvent のプロパティ

event = CloudEvent(
    type="MyApp.Events.ItemCreated",      # Required: event type
    source="/myapp/items",                 # Required: event source
    data={"key": "value"},                 # Event payload
    subject="items/123",                   # Optional: subject/path
    datacontenttype="application/json",   # Optional: content type
    dataschema="https://schema.example",  # Optional: schema URL
    time=datetime.now(timezone.utc),      # Optional: timestamp
    extensions={"custom": "value"}         # Optional: custom attributes
)

EventGridEvent のプロパティ

event = EventGridEvent(
    subject="/myapp/items/123",            # Required: subject
    event_type="MyApp.ItemCreated",        # Required: event type
    data={"key": "value"},                 # Required: event payload
    data_version="1.0",                    # Required: schema version
    topic="/subscriptions/.../topics/...", # Optional: auto-set
    event_time=datetime.now(timezone.utc)  # Optional: timestamp
)

非同期クライアント

from azure.eventgrid.aio import EventGridPublisherClient
from azure.identity.aio import DefaultAzureCredential

async def publish_events():
    async with DefaultAzureCredential() as credential, \
            EventGridPublisherClient(endpoint, credential) as client:
        event = CloudEvent(
            type="MyApp.Events.Test",
            source="/myapp",
            data={"message": "hello"}
        )
        await client.send(event)

import asyncio
asyncio.run(publish_events())

名前空間トピック（Event Grid Namespaces）

Event Grid Namespaces（プル配信）の場合：

from azure.eventgrid.aio import EventGridPublisherClient

# Namespace endpoint (different from custom topic)
namespace_endpoint = "https://<namespace>.<region>.eventgrid.azure.net"
topic_name = "my-topic"

async with EventGridPublisherClient(
    endpoint=namespace_endpoint,
    credential=credential,  # an open azure.identity.aio credential
    namespace_topic=topic_name,
) as client:
    await client.send(event)

名前空間トピックは CloudEvents をサポートしています。名前空間クライアントは namespace_topic を指定して構築してください。送信専用のキーワードとして使用しないでください。使用している SDK のバージョンに合った例を参照してください。テスト用トピックと合成ペイロードを使用してください。発行によって実際のサブスクライバーが起動する可能性があります。認証にはスコープが設定されたデータ送信者ロールが必要です。また、アプリケーションのコンシューマーは重複配信や冪等性に対応する必要があります。非同期クライアントと認証情報の両方をクローズしてください。公式 Python SDK ガイドを参照してください。

ベストプラクティス

1. 新しいアプリケーションではCloudEvents を使用します（業界標準）。
2. 複数のイベントを発行する場合はイベントをバッチ処理します。
3. フィルタリングに役立つ意味のあるサブジェクトを含めます。
4. 高スループットのシナリオでは非同期クライアントを使用します。
5. 再試行を処理します — Event Grid には再試行機能が組み込まれています。
6. ルーティングとフィルタリングに適したイベントの種類を設定します。

使用する場合

このスキルは、概要に記載されたワークフローまたはアクションを実行する場合に適用できます。

制限事項

• このスキルは、タスクが上記の説明の範囲に明確に合致する場合にのみ使用してください。
• 出力を、環境固有の検証、テスト、専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、停止して確認してください。