azure-eventgrid-py

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Azure Event Grid SDK for Python

适用于Python的Azure Event Grid SDK

Event routing service for building event-driven applications with pub/sub semantics.

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

Installation

安装

bash

pip install azure-eventgrid azure-identity

bash

pip install azure-eventgrid azure-identity

Environment Variables

环境变量

bash

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

bash

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

Authentication

身份验证

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)

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)

Event Types

事件类型

Format	Class	Use Case
Cloud Events 1.0	`CloudEvent`	Standard, interoperable (recommended)
Event Grid Schema	`EventGridEvent`	Azure-native format

格式	类	使用场景
Cloud Events 1.0	`CloudEvent`	标准、可互操作（推荐）
Event Grid Schema	`EventGridEvent`	Azure原生格式

Publish CloudEvents

发布CloudEvents

python

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

client = EventGridPublisherClient(endpoint, DefaultAzureCredential())

python

from azure.eventgrid import EventGridPublisherClient, 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)

undefined

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

undefined

Publish EventGridEvents

发布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)

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)

Event Properties

事件属性

CloudEvent Properties

CloudEvent属性

python

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
)

python

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 Properties

EventGridEvent属性

python

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
)

python

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
)

Async Client

异步客户端

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.Test",
            source="/myapp",
            data={"message": "hello"}
        )
        await client.send(event)

import asyncio
asyncio.run(publish_events())

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.Test",
            source="/myapp",
            data={"message": "hello"}
        )
        await client.send(event)

import asyncio
asyncio.run(publish_events())

Namespace Topics (Event Grid Namespaces)

命名空间主题（Event Grid命名空间）

For Event Grid Namespaces (pull delivery):

python

from azure.eventgrid.aio import EventGridPublisherClient

对于Event Grid命名空间（拉取式交付）：

python

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=DefaultAzureCredential() ) as client: await client.send( event, namespace_topic=topic_name )

undefined

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 )

undefined

Best Practices

最佳实践

Use CloudEvents for new applications (industry standard)
Batch events when publishing multiple events
Include meaningful subjects for filtering
Use async client for high-throughput scenarios
Handle retries — Event Grid has built-in retry
Set appropriate event types for routing and filtering

使用CloudEvents 用于新应用程序（行业标准）
批量发布事件 当发布多个事件时
包含有意义的主题 用于过滤
使用异步客户端 用于高吞吐量场景
处理重试 —— Event Grid内置重试机制
设置合适的事件类型 用于路由和过滤