Azure Event Grid Python SDK

azure-eventgrid-py
分类编程
作者Agentic Awesome Skills 社区
许可MIT
评分4.60/5
使用13.6K

Azure Event Grid Python SDK

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

安装

bash
pip install azure-eventgrid azure-identity

环境变量

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

身份验证

python
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

python
from azure.eventgrid import EventGridPublisherClient, CloudEvent
from azure.identity import DefaultAzureCredential

client = EventGridPublisherClient(endpoint, DefaultAzureCredential())


单个事件

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

多个事件

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

发布 EventGridEvents

python
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 属性

python
event = CloudEvent(
    type="MyApp.Events.ItemCreated",      # 必填：事件类型
    source="/myapp/items",                 # 必填：事件源
    data={"key": "value"},                 # 事件负载
    subject="items/123",                   # 可选：主题/路径
    datacontenttype="application/json",   # 可选：内容类型
    dataschema="https://schema.example",  # 可选：Schema URL
    time=datetime.now(timezone.utc),      # 可选：时间戳
    extensions={"custom": "value"}         # 可选：自定义属性
)

EventGridEvent 属性

python
event = EventGridEvent(
    subject="/myapp/items/123",            # 必填：主题
    event_type="MyApp.ItemCreated",        # 必填：事件类型
    data={"key": "value"},                 # 必填：事件负载
    data_version="1.0",                    # 必填：Schema 版本
    topic="/subscriptions/.../topics/...", # 可选：自动设置
    event_time=datetime.now(timezone.utc)  # 可选：时间戳
)

异步客户端

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

async def publish_events():
    credential = DefaultAzureCredential()
    
    async with EventGridPublisherClient(endpoint, credential) as client:
        event = CloudEvent(
            type="MyApp.Events.T

est",
source="/myapp",
data={"message": "hello"}
)
await client.send(event)

import asyncio
asyncio.run(publish_events())

code
## 命名空间主题 (Event Grid Namespaces)


对于 Event Grid 命名空间（拉取模式）：
python
from azure.eventgrid.aio import EventGridPublisherClient

命名空间终结点（与自定义主题不同）

namespace_endpoint = "https://<namespace>.<region>.eventgrid.azure.net" topic_name = "my-topic"

async with EventGridPublisherClient(
endpoint=namespace_endpoint,
credential=DefaultAzureCredential()
) as client:
await client.send(
event,
namespace_topic=topic_name
)
```

最佳实践

1. 新应用优先使用 CloudEvents（行业标准）
2. 批量发送事件：在发布多个事件时采用批处理
3. 包含有意义的主题 (Subject)：以便于过滤
4. 高吞吐量场景使用异步客户端 (async client)
5. 处理重试：Event Grid 具有内置重试机制
6. 设置合适的事件类型：用于路由和过滤

适用场景

此技能适用于执行概述中描述的工作流或操作。

局限性

  • 仅在任务与上述范围明确匹配时使用此技能。
  • 不要将输出视为针对特定环境的验证、测试或专家评审的替代方案。
  • 如果缺少必要的输入、权限、安全边界或成功标准，请停止并请求澄清。