azure-eventgrid-py

Azure Event Grid SDK for Python

用于构建具有发布/订阅语义的事件驱动型应用程序的事件路由服务。

安装

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 Schema	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 命名空间）

适用于 Event Grid 命名空间（拉取式传递）：

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. 设置合适的事件类型以进行路由和筛选

适用场景

当任务与概述中描述的工作流或操作相符时，可使用此技能。

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 请勿将输出视为环境特定验证、测试或专家审查的替代品。
• 如果缺少必要的输入、权限、安全边界或成功标准，请暂停并请求澄清。