Azure Eventgrid Py

1---
2name: azure-eventgrid-py
3description: |
4  Azure Event Grid SDK for Python. Use for publishing events, handling CloudEvents, and event-driven architectures.
5  Triggers: "event grid", "EventGridPublisherClient", "CloudEvent", "EventGridEvent", "publish events".
6package: azure-eventgrid
7---
8 
9# Azure Event Grid SDK for Python
10 
11Event routing service for building event-driven applications with pub/sub semantics.
12 
13## Installation
14 
15```bash
16pip install azure-eventgrid azure-identity
17```
18 
19## Environment Variables
20 
21```bash
22EVENTGRID_TOPIC_ENDPOINT=https://<topic-name>.<region>.eventgrid.azure.net/api/events
23EVENTGRID_NAMESPACE_ENDPOINT=https://<namespace>.<region>.eventgrid.azure.net
24```
25 
26## Authentication
27 
28```python
29from azure.identity import DefaultAzureCredential
30from azure.eventgrid import EventGridPublisherClient
31 
32credential = DefaultAzureCredential()
33endpoint = "https://<topic-name>.<region>.eventgrid.azure.net/api/events"
34 
35client = EventGridPublisherClient(endpoint, credential)
36```
37 
38## Event Types
39 
40| Format | Class | Use Case |
41|--------|-------|----------|
42| Cloud Events 1.0 | `CloudEvent` | Standard, interoperable (recommended) |
43| Event Grid Schema | `EventGridEvent` | Azure-native format |
44 
45## Publish CloudEvents
46 
47```python
48from azure.eventgrid import EventGridPublisherClient, CloudEvent
49from azure.identity import DefaultAzureCredential
50 
51client = EventGridPublisherClient(endpoint, DefaultAzureCredential())
52 
53# Single event
54event = CloudEvent(
55    type="MyApp.Events.OrderCreated",
56    source="/myapp/orders",
57    data={"order_id": "12345", "amount": 99.99}
58)
59client.send(event)
60 
61# Multiple events
62events = [
63    CloudEvent(
64        type="MyApp.Events.OrderCreated",
65        source="/myapp/orders",
66        data={"order_id": f"order-{i}"}
67    )
68    for i in range(10)
69]
70client.send(events)
71```
72 
73## Publish EventGridEvents
74 
75```python
76from azure.eventgrid import EventGridEvent
77from datetime import datetime, timezone
78 
79event = EventGridEvent(
80    subject="/myapp/orders/12345",
81    event_type="MyApp.Events.OrderCreated",
82    data={"order_id": "12345", "amount": 99.99},
83    data_version="1.0"
84)
85 
86client.send(event)
87```
88 
89## Event Properties
90 
91### CloudEvent Properties
92 
93```python
94event = CloudEvent(
95    type="MyApp.Events.ItemCreated",      # Required: event type
96    source="/myapp/items",                 # Required: event source
97    data={"key": "value"},                 # Event payload
98    subject="items/123",                   # Optional: subject/path
99    datacontenttype="application/json",   # Optional: content type
100    dataschema="https://schema.example",  # Optional: schema URL
101    time=datetime.now(timezone.utc),      # Optional: timestamp
102    extensions={"custom": "value"}         # Optional: custom attributes
103)
104```
105 
106### EventGridEvent Properties
107 
108```python
109event = EventGridEvent(
110    subject="/myapp/items/123",            # Required: subject
111    event_type="MyApp.ItemCreated",        # Required: event type
112    data={"key": "value"},                 # Required: event payload
113    data_version="1.0",                    # Required: schema version
114    topic="/subscriptions/.../topics/...", # Optional: auto-set
115    event_time=datetime.now(timezone.utc)  # Optional: timestamp
116)
117```
118 
119## Async Client
120 
121```python
122from azure.eventgrid.aio import EventGridPublisherClient
123from azure.identity.aio import DefaultAzureCredential
124 
125async def publish_events():
126    credential = DefaultAzureCredential()
127    
128    async with EventGridPublisherClient(endpoint, credential) as client:
129        event = CloudEvent(
130            type="MyApp.Events.Test",
131            source="/myapp",
132            data={"message": "hello"}
133        )
134        await client.send(event)
135 
136import asyncio
137asyncio.run(publish_events())
138```
139 
140## Namespace Topics (Event Grid Namespaces)
141 
142For Event Grid Namespaces (pull delivery):
143 
144```python
145from azure.eventgrid.aio import EventGridPublisherClient
146 
147# Namespace endpoint (different from custom topic)
148namespace_endpoint = "https://<namespace>.<region>.eventgrid.azure.net"
149topic_name = "my-topic"
150 
151async with EventGridPublisherClient(
152    endpoint=namespace_endpoint,
153    credential=DefaultAzureCredential()
154) as client:
155    await client.send(
156        event,
157        namespace_topic=topic_name
158    )
159```
160 
161## Best Practices
162 
1631. **Use CloudEvents** for new applications (industry standard)
1642. **Batch events** when publishing multiple events
1653. **Include meaningful subjects** for filtering
1664. **Use async client** for high-throughput scenarios
1675. **Handle retries** — Event Grid has built-in retry
1686. **Set appropriate event types** for routing and filtering
169