0
# Legacy Support
1

2
Azure Event Grid Python SDK provides comprehensive backward compatibility for Event Grid Basic (legacy) scenarios. This includes EventGrid schema events, system event names, SAS token generation, and legacy publisher clients.
3

4
## Capabilities
5

6
### EventGrid Schema Events
7

8
Traditional EventGrid event format for Event Grid Basic topics and domains.
9

10
```python { .api }
11
class EventGridEvent:
12
    def __init__(
13
        self,
14
        subject: str,
15
        event_type: str,
16
        data: Any,
17
        data_version: str,
18
        *,
19
        topic: Optional[str] = None,
20
        metadata_version: Optional[str] = None,
21
        id: Optional[str] = None,
22
        event_time: Optional[datetime] = None
23
    ) -> None:
24
        """
25
        Create EventGrid schema event.
26

27
        Parameters:
28
        - subject: Subject of the event in the context of the event producer (required)
29
        - event_type: One of the registered event types for this event source (required)  
30
        - data: Event data specific to the resource provider (required)
31
        - data_version: The schema version of the data object (required)
32
        - topic: The resource path to the event source (optional, set by service)
33
        - metadata_version: The schema version of the event metadata (optional)
34
        - id: An identifier for the event (optional, auto-generated if not provided)
35
        - event_time: The time the event was generated (optional, auto-generated if not provided)
36
        """
37

38
    @classmethod
39
    def from_json(cls, event: Union[str, dict]) -> "EventGridEvent":
40
        """
41
        Create EventGridEvent from JSON string or dictionary.
42

43
        Parameters:
44
        - event: JSON string or dictionary representing the event
45

46
        Returns:
47
        EventGridEvent: Parsed event object
48
        """
49
```
50

51
### System Event Names
52

53
Comprehensive enumeration of Azure system event types for filtering and handling built-in Azure service events.
54

55
```python { .api }
56
class SystemEventNames(str, Enum):
57
    """
58
    Enumeration of Azure system event names.
59
    
60
    Contains 100+ system event types including:
61
    - EventGridSubscriptionValidationEventName
62
    - ResourceWriteSuccessEventName  
63
    - ResourceWriteFailureEventName
64
    - ResourceDeleteSuccessEventName
65
    - IoTHubDeviceConnectedEventName
66
    - IoTHubDeviceDisconnectedEventName
67
    - StorageBlobCreatedEventName
68
    - StorageBlobDeletedEventName
69
    - ServiceBusActiveMessagesAvailableWithNoListenersEventName
70
    - And many more...
71
    """
72
    
73
    # Event Grid events
74
    EventGridSubscriptionValidationEventName = "Microsoft.EventGrid.SubscriptionValidationEvent"
75
    EventGridSubscriptionDeletedEventName = "Microsoft.EventGrid.SubscriptionDeletedEvent"
76
    
77
    # Resource Manager events  
78
    ResourceWriteSuccessEventName = "Microsoft.Resources.ResourceWriteSuccess"
79
    ResourceWriteFailureEventName = "Microsoft.Resources.ResourceWriteFailure"
80
    ResourceWriteCancelEventName = "Microsoft.Resources.ResourceWriteCancel"
81
    ResourceDeleteSuccessEventName = "Microsoft.Resources.ResourceDeleteSuccess"
82
    ResourceDeleteFailureEventName = "Microsoft.Resources.ResourceDeleteFailure"
83
    ResourceDeleteCancelEventName = "Microsoft.Resources.ResourceDeleteCancel"
84
    
85
    # Storage events
86
    StorageBlobCreatedEventName = "Microsoft.Storage.BlobCreated"
87
    StorageBlobDeletedEventName = "Microsoft.Storage.BlobDeleted"
88
    StorageBlobRenamedEventName = "Microsoft.Storage.BlobRenamed"
89
    StorageDirectoryCreatedEventName = "Microsoft.Storage.DirectoryCreated"
90
    StorageDirectoryDeletedEventName = "Microsoft.Storage.DirectoryDeleted"
91
    StorageDirectoryRenamedEventName = "Microsoft.Storage.DirectoryRenamed"
92
    
93
    # IoT Hub events
94
    IoTHubDeviceConnectedEventName = "Microsoft.Devices.DeviceConnected"
95
    IoTHubDeviceDisconnectedEventName = "Microsoft.Devices.DeviceDisconnected"
96
    IoTHubDeviceCreatedEventName = "Microsoft.Devices.DeviceCreated"
97
    IoTHubDeviceDeletedEventName = "Microsoft.Devices.DeviceDeleted"
98
    IoTHubDeviceTelemetryEventName = "Microsoft.Devices.DeviceTelemetry"
99
    
100
    # And 90+ more system event names...
101
```
102

103
### SAS Token Generation
104

105
Generate Shared Access Signature tokens for Event Grid Basic authentication.
106

107
```python { .api }
108
def generate_sas(
109
    endpoint: str,
110
    shared_access_key: str,
111
    expiration_date_utc: datetime,
112
    *,
113
    api_version: Optional[str] = None
114
) -> str:
115
    """
116
    Generate Shared Access Signature for Event Grid authentication.
117

118
    Parameters:
119
    - endpoint: Event Grid topic endpoint URL
120
    - shared_access_key: Shared access key for the topic
121
    - expiration_date_utc: UTC datetime when the token expires
122
    - api_version: API version for the signature (optional)
123

124
    Returns:
125
    str: SAS token for authentication
126

127
    Raises:
128
    - ValueError: Invalid endpoint or key format
129
    - TypeError: Invalid expiration date type
130
    """
131
```
132

133
### Legacy Async Publisher
134

135
Asynchronous Event Grid Basic publisher for backward compatibility.
136

137
```python { .api }
138
# Available in azure.eventgrid._legacy.aio
139
class EventGridPublisherClient:
140
    def __init__(
141
        self,
142
        endpoint: str,
143
        credential: Union[AzureKeyCredential, AzureSasCredential],
144
        **kwargs: Any
145
    ) -> None: ...
146

147
    async def send(
148
        self,
149
        events: Union[EventGridEvent, List[EventGridEvent]],
150
        **kwargs: Any
151
    ) -> None:
152
        """
153
        Asynchronously send EventGrid schema events to Event Grid Basic.
154

155
        Parameters:
156
        - events: Single event or list of EventGrid events
157
        """
158
```
159

160
## Usage Examples
161

162
### Event Grid Basic Publishing
163

164
```python
165
from azure.eventgrid import EventGridPublisherClient, EventGridEvent
166
from azure.core.credentials import AzureKeyCredential
167
from datetime import datetime
168

169
# Create publisher for Event Grid Basic topic
170
publisher = EventGridPublisherClient(
171
    endpoint="https://my-topic.westus.eventgrid.azure.net/api/events",
172
    credential=AzureKeyCredential("access_key")
173
    # Note: No namespace_topic parameter for Event Grid Basic
174
)
175

176
# Create EventGrid schema event
177
event = EventGridEvent(
178
    subject="orders/order-12345",
179
    event_type="Contoso.Orders.OrderCreated",
180
    data={
181
        "order_id": "12345",
182
        "customer_id": "customer-67890",
183
        "product_ids": ["product-1", "product-2"],
184
        "total_amount": 149.99,
185
        "currency": "USD"
186
    },
187
    data_version="1.0"
188
)
189

190
# Send single event
191
publisher.send(event)
192

193
# Send multiple events
194
events = [
195
    EventGridEvent(
196
        subject="orders/order-12346",
197
        event_type="Contoso.Orders.OrderCreated", 
198
        data={"order_id": "12346", "total": 99.99},
199
        data_version="1.0"
200
    ),
201
    EventGridEvent(
202
        subject="orders/order-12347",
203
        event_type="Contoso.Orders.OrderCreated",
204
        data={"order_id": "12347", "total": 199.99}, 
205
        data_version="1.0"
206
    )
207
]
208

209
publisher.send(events)
210
publisher.close()
211
```
212

213
### Creating Events with All Properties
214

215
```python
216
from azure.eventgrid import EventGridEvent
217
from datetime import datetime, timezone
218

219
# Event with all optional properties
220
detailed_event = EventGridEvent(
221
    subject="products/electronics/laptops/laptop-abc123",
222
    event_type="Contoso.Inventory.ProductUpdated",
223
    data={
224
        "product_id": "laptop-abc123",
225
        "name": "Gaming Laptop Pro",
226
        "category": "electronics/laptops",
227
        "price": 1299.99,
228
        "inventory_count": 15,
229
        "updated_fields": ["price", "inventory_count"]
230
    },
231
    data_version="2.1",
232
    id="event-unique-id-789",
233
    event_time=datetime.now(timezone.utc),
234
    topic="contoso-inventory",  # Usually set by the service
235
    metadata_version="1"
236
)
237

238
publisher.send(detailed_event)
239
```
240

241
### Working with System Events
242

243
```python
244
from azure.eventgrid import SystemEventNames
245

246
# Filter events by system event types
247
def handle_system_event(event_type, event_data):
248
    """Handle different types of Azure system events."""
249
    
250
    if event_type == SystemEventNames.StorageBlobCreatedEventName:
251
        blob_url = event_data.get('url')
252
        print(f"New blob created: {blob_url}")
253
        
254
    elif event_type == SystemEventNames.ResourceWriteSuccessEventName:
255
        resource_uri = event_data.get('resourceUri')
256
        operation = event_data.get('operationName')
257
        print(f"Resource operation successful: {operation} on {resource_uri}")
258
        
259
    elif event_type == SystemEventNames.IoTHubDeviceConnectedEventName:
260
        device_id = event_data.get('deviceId')
261
        print(f"IoT device connected: {device_id}")
262
        
263
    elif event_type == SystemEventNames.EventGridSubscriptionValidationEventName:
264
        validation_code = event_data.get('validationCode')
265
        validation_url = event_data.get('validationUrl')
266
        print(f"Subscription validation required: {validation_code}")
267
        # Handle validation logic
268
        
269
    else:
270
        print(f"Unknown system event: {event_type}")
271

272
# Example system event handling
273
system_events = [
274
    {
275
        "eventType": SystemEventNames.StorageBlobCreatedEventName,
276
        "data": {
277
            "url": "https://mystorageaccount.blob.core.windows.net/container/newfile.txt",
278
            "contentType": "text/plain",
279
            "contentLength": 1024
280
        }
281
    }
282
]
283

284
for event in system_events:
285
    handle_system_event(event["eventType"], event["data"])
286
```
287

288
### SAS Authentication
289

290
```python
291
from azure.eventgrid import generate_sas
292
from azure.core.credentials import AzureSasCredential
293
from datetime import datetime, timedelta
294

295
# Generate SAS token
296
endpoint = "https://my-topic.westus.eventgrid.azure.net/api/events"
297
access_key = "your_shared_access_key_here"
298
expiration = datetime.utcnow() + timedelta(hours=2)
299

300
sas_token = generate_sas(
301
    endpoint=endpoint,
302
    shared_access_key=access_key,
303
    expiration_date_utc=expiration,
304
    api_version="2018-01-01"
305
)
306

307
print(f"Generated SAS token: {sas_token}")
308

309
# Use SAS token for authentication
310
publisher = EventGridPublisherClient(
311
    endpoint=endpoint,
312
    credential=AzureSasCredential(sas_token)
313
)
314

315
# Create and send event
316
event = EventGridEvent(
317
    subject="auth/sas-example",
318
    event_type="Example.SasAuth",
319
    data={"message": "Authenticated with SAS token"},
320
    data_version="1.0"
321
)
322

323
publisher.send(event)
324
publisher.close()
325
```
326

327
### Creating Events from JSON
328

329
```python
330
import json
331
from azure.eventgrid import EventGridEvent
332

333
# JSON string representation
334
json_event = """
335
{
336
    "id": "event-123",
337
    "subject": "myapp/users/user456",  
338
    "eventType": "MyApp.User.ProfileUpdated",
339
    "eventTime": "2024-01-15T10:30:00Z",
340
    "data": {
341
        "userId": "user456",
342
        "updatedFields": ["email", "phoneNumber"],
343
        "previousEmail": "old@example.com",
344
        "newEmail": "new@example.com"
345
    },
346
    "dataVersion": "1.0"
347
}
348
"""
349

350
# Create event from JSON string
351
event = EventGridEvent.from_json(json_event)
352
publisher.send(event)
353

354
# Create event from dictionary
355
event_dict = {
356
    "subject": "orders/order-789",
357
    "eventType": "Contoso.Orders.OrderShipped", 
358
    "data": {
359
        "orderId": "789",
360
        "trackingNumber": "TRACK123456",
361
        "estimatedDelivery": "2024-01-20"
362
    },
363
    "dataVersion": "1.0"
364
}
365

366
event = EventGridEvent.from_json(event_dict)
367
publisher.send(event)
368
```
369

370
### Legacy Async Publishing
371

372
```python
373
import asyncio
374
from azure.eventgrid._legacy.aio import EventGridPublisherClient
375
from azure.eventgrid import EventGridEvent
376
from azure.core.credentials import AzureKeyCredential
377

378
async def async_legacy_publishing():
379
    """Example of async Event Grid Basic publishing."""
380
    
381
    # Create async legacy publisher
382
    async with EventGridPublisherClient(
383
        endpoint="https://my-topic.westus.eventgrid.azure.net/api/events",
384
        credential=AzureKeyCredential("access_key")
385
    ) as publisher:
386
        
387
        # Create events
388
        events = [
389
            EventGridEvent(
390
                subject=f"async/event-{i}",
391
                event_type="AsyncExample.EventCreated",
392
                data={"index": i, "timestamp": datetime.now().isoformat()},
393
                data_version="1.0"
394
            )
395
            for i in range(5)
396
        ]
397
        
398
        # Send events asynchronously
399
        await publisher.send(events)
400
        print(f"Sent {len(events)} events asynchronously")
401

402
# Run async function
403
asyncio.run(async_legacy_publishing())
404
```
405

406
### Event Validation Patterns
407

408
```python
409
from azure.eventgrid import EventGridEvent, SystemEventNames
410

411
def validate_eventgrid_event(event_data):
412
    """Validate EventGrid event structure and required fields."""
413
    
414
    required_fields = ['subject', 'eventType', 'data', 'dataVersion']
415
    
416
    for field in required_fields:
417
        if field not in event_data:
418
            raise ValueError(f"Missing required field: {field}")
419
    
420
    # Validate event type format
421
    event_type = event_data['eventType']
422
    if not event_type or '.' not in event_type:
423
        raise ValueError("eventType should follow format: Publisher.Object.Action")
424
    
425
    # Validate subject format
426
    subject = event_data['subject']
427
    if not subject or not subject.startswith('/'):
428
        # Subject should be a resource path
429
        pass  # Flexible validation
430
    
431
    # Validate data version
432
    data_version = event_data['dataVersion']
433
    if not data_version:
434
        raise ValueError("dataVersion is required")
435
    
436
    return True
437

438
# Example validation
439
event_data = {
440
    "subject": "/orders/order-123",
441
    "eventType": "Contoso.Orders.OrderProcessed",
442
    "data": {"orderId": "123", "status": "completed"},
443
    "dataVersion": "1.0"
444
}
445

446
try:
447
    validate_eventgrid_event(event_data)
448
    event = EventGridEvent.from_json(event_data)
449
    print("Event validation successful")
450
except ValueError as e:
451
    print(f"Event validation failed: {e}")
452
```
453

454
### Event Grid Basic vs Namespaces Comparison
455

456
```python
457
from azure.eventgrid import EventGridPublisherClient, EventGridEvent
458
from azure.core.credentials import AzureKeyCredential
459
from azure.core.messaging import CloudEvent
460

461
# Event Grid Basic (Legacy)
462
basic_publisher = EventGridPublisherClient(
463
    endpoint="https://topic.region.eventgrid.azure.net/api/events",
464
    credential=AzureKeyCredential("key")
465
    # No namespace_topic specified = Event Grid Basic
466
)
467

468
eventgrid_event = EventGridEvent(
469
    subject="legacy/example",
470
    event_type="Legacy.EventType",
471
    data={"key": "value"},
472
    data_version="1.0"
473
)
474

475
basic_publisher.send(eventgrid_event)
476

477
# Event Grid Namespaces (Modern)  
478
namespace_publisher = EventGridPublisherClient(
479
    endpoint="https://namespace.region.eventgrid.azure.net",
480
    credential=AzureKeyCredential("key"),
481
    namespace_topic="my-topic"  # Specifying namespace_topic = Event Grid Namespaces
482
)
483

484
cloud_event = CloudEvent(
485
    source="modern/example",
486
    type="Modern.EventType", 
487
    data={"key": "value"}
488
)
489

490
namespace_publisher.send(cloud_event)
491

492
# Clean up
493
basic_publisher.close()
494
namespace_publisher.close()
495
```
496

497
### Error Handling for Legacy Operations
498

499
```python
500
from azure.core.exceptions import HttpResponseError, ClientAuthenticationError
501
from azure.eventgrid import generate_sas
502
from datetime import datetime, timedelta
503

504
def robust_legacy_publishing(endpoint, access_key, events):
505
    """Robust Event Grid Basic publishing with error handling."""
506
    
507
    try:
508
        # Generate SAS token with reasonable expiration
509
        expiration = datetime.utcnow() + timedelta(hours=1)
510
        sas_token = generate_sas(endpoint, access_key, expiration)
511
        
512
        # Create publisher with SAS authentication
513
        publisher = EventGridPublisherClient(
514
            endpoint=endpoint,
515
            credential=AzureSasCredential(sas_token)
516
        )
517
        
518
        # Send events with retry logic
519
        max_retries = 3
520
        for attempt in range(max_retries):
521
            try:
522
                publisher.send(events)
523
                print(f"Successfully sent {len(events)} events")
524
                break
525
                
526
            except HttpResponseError as e:
527
                if e.status_code == 401:
528
                    # Authentication failed - regenerate token
529
                    print("Authentication failed, regenerating SAS token...")
530
                    new_expiration = datetime.utcnow() + timedelta(hours=1)
531
                    new_sas_token = generate_sas(endpoint, access_key, new_expiration)
532
                    publisher = EventGridPublisherClient(
533
                        endpoint=endpoint,
534
                        credential=AzureSasCredential(new_sas_token)
535
                    )
536
                    
537
                elif e.status_code == 413:
538
                    # Payload too large - split batch
539
                    print("Payload too large, splitting batch...")
540
                    mid = len(events) // 2
541
                    robust_legacy_publishing(endpoint, access_key, events[:mid])
542
                    robust_legacy_publishing(endpoint, access_key, events[mid:])
543
                    return
544
                    
545
                elif attempt == max_retries - 1:
546
                    print(f"Failed after {max_retries} attempts: {e}")
547
                    raise
548
                else:
549
                    print(f"Attempt {attempt + 1} failed, retrying...")
550
                    time.sleep(2 ** attempt)  # Exponential backoff
551
                    
552
        publisher.close()
553
        
554
    except ValueError as e:
555
        print(f"Invalid parameters: {e}")
556
        raise
557
    except Exception as e:
558
        print(f"Unexpected error: {e}")
559
        raise
560

561
# Usage
562
events = [
563
    EventGridEvent(
564
        subject="robust/example",
565
        event_type="Example.RobustPublishing",
566
        data={"attempt": i},
567
        data_version="1.0"
568
    )
569
    for i in range(10)
570
]
571

572
robust_legacy_publishing(
573
    endpoint="https://topic.region.eventgrid.azure.net/api/events",
574
    access_key="your_access_key",
575
    events=events
576
)
577
```