Azure Event Grid Client Library

Warnings

• breaking The `azure-eventgrid` library underwent significant API changes between its V1 (1.x.x) and V4 (4.x.x) versions. Classes like `EventGridClient` for publishing and the `azure.eventgrid.models` submodule for event types (`EventGridEvent`, `CloudEvent`) were replaced.
  Fix: Upgrade to `azure-eventgrid>=4.0.0` and update your code. Use `EventGridPublisherClient` for publishing and import `EventGridEvent` or `CloudEvent` directly from `azure.eventgrid`.
• gotcha Event Grid supports two main event schemas: CloudEvents 1.0 and the Event Grid schema. You must correctly choose and construct the event type (`CloudEvent` or `EventGridEvent`) corresponding to the schema your Event Grid topic is configured to accept, or that you expect to receive. Mismatched schemas can lead to events being dropped or malformed.
  Fix: Carefully review your Event Grid topic's schema configuration and use the appropriate `CloudEvent` or `EventGridEvent` class. When publishing, the client will serialize events according to the object type provided. When deserializing, `EventGridDeserializer` can handle both.
• gotcha There are two distinct Python libraries: `azure-eventgrid` (this data plane client) and `azure-mgmt-eventgrid` (the management plane client). `azure-eventgrid` is for sending/receiving actual events, while `azure-mgmt-eventgrid` is for creating, updating, and deleting Event Grid topics, domains, and subscriptions. Using the wrong library for your task will result in `AttributeError` or unexpected behavior.
  Fix: Ensure you import from `azure.eventgrid` for data plane operations (publishing/deserializing events) and `azure.mgmt.eventgrid` for resource management operations.
• gotcha When receiving events via HTTP (e.g., a webhook), Event Grid sends a 'webhook validation' event. Your application must respond to this validation event within 30 seconds by returning the 'validationCode' in the response body. Failing to do so will prevent the subscription from becoming active.
  Fix: Implement logic in your webhook endpoint to detect 'Microsoft.EventGrid.SubscriptionValidationEvent' and return a JSON response like `{'validationResponse': event.data['validationCode']}`.

Install

• pip install azure-eventgrid Install stable release

Imports

• EventGridPublisherClient

  from azure.eventgrid import EventGridPublisherClient

  EventGridClient was used in older V1 versions; EventGridPublisherClient is for current V4+.
• EventGridDeserializer

  from azure.eventgrid import EventGridDeserializer

• CloudEvent

  from azure.eventgrid import CloudEvent

  In V4+, EventGridEvent and CloudEvent are directly under azure.eventgrid, not a 'models' submodule.
• EventGridEvent

  from azure.eventgrid import EventGridEvent

  In V4+, EventGridEvent and CloudEvent are directly under azure.eventgrid, not a 'models' submodule.
• AzureKeyCredential

  from azure.core.credentials import AzureKeyCredential

Quickstart

This quickstart demonstrates how to publish both CloudEvents and Event Grid Schema events to an Event Grid topic using `EventGridPublisherClient` and then how to deserialize incoming event payloads using `EventGridDeserializer`. Ensure `EVENT_GRID_ENDPOINT` and `EVENT_GRID_KEY` environment variables are set.

import os
import json
from azure.eventgrid import EventGridPublisherClient, EventGridEvent, CloudEvent, EventGridDeserializer
from azure.core.credentials import AzureKeyCredential

# Set your Event Grid endpoint and access key from environment variables
EVENT_GRID_ENDPOINT = os.environ.get("EVENT_GRID_ENDPOINT", "")
EVENT_GRID_KEY = os.environ.get("EVENT_GRID_KEY", "")

if not EVENT_GRID_ENDPOINT or not EVENT_GRID_KEY:
    print("Please set the environment variables EVENT_GRID_ENDPOINT and EVENT_GRID_KEY.")
    exit(1)

credential = AzureKeyCredential(EVENT_GRID_KEY)
client = EventGridPublisherClient(EVENT_GRID_ENDPOINT, credential)

# --- Publish a CloudEvent ---
cloud_event = CloudEvent(
    source="https://example.com/myapp",
    type="Contoso.Items.ItemReceived",
    data={"itemSku": "CONTOSO-ITEM-0001", "itemCount": 1},
    subject="MySubject",
    # Additional CloudEvent attributes can be passed as keyword arguments
)
client.send([cloud_event])
print("Published a CloudEvent successfully.")

# --- Publish an EventGridEvent ---
event_grid_event = EventGridEvent(
    subject="Contoso/Items/ItemReceived",
    data={"itemSku": "CONTOSO-ITEM-0002", "itemCount": 1},
    event_type="Contoso.Items.ItemReceived",
    data_version="1.0",
)
client.send([event_grid_event])
print("Published an EventGridEvent successfully.")

# --- Deserialize an incoming event (e.g., from a webhook payload) ---
simulated_webhook_payload = json.dumps([
    {
        "id": "test-event-id-1",
        "source": "/azure/events/test",
        "data": {"message": "Hello CloudEvent!"},
        "type": "test.event",
        "time": "2023-10-27T10:00:00Z",
        "specversion": "1.0"
    },
    {
        "id": "test-event-id-2",
        "topic": "/subscriptions/xxx/resourceGroups/yyy/providers/Microsoft.EventGrid/topics/mytesttopic",
        "subject": "TestSubject",
        "data": {"message": "Hello EventGridEvent!"},
        "eventTime": "2023-10-27T10:05:00Z",
        "eventType": "Microsoft.Storage.BlobCreated",
        "dataVersion": "1.0",
        "metadataVersion": "1",
        "blobType": "BlockBlob"
    }
])

deserializer = EventGridDeserializer()
deserialized_events = deserializer.deserialize_events(simulated_webhook_payload)

print("\nDeserialized events from simulated webhook payload:")
for event in deserialized_events:
    if isinstance(event, CloudEvent):
        print(f"  CloudEvent - ID: {event.id}, Type: {event.type}, Data: {event.data}")
    elif isinstance(event, EventGridEvent):
        print(f"  EventGridEvent - ID: {event.id}, Type: {event.event_type}, Subject: {event.subject}")