SDK Azure Event Grid pour Python

Service de routage d'événements pour construire des applications pilotées par les événements avec une sémantique pub/sub.

Installation

pip install azure-eventgrid azure-identity

Variables d'environnement

EVENTGRID_TOPIC_ENDPOINT=https://<topic-name>.<region>.eventgrid.azure.net/api/events  # Requis pour la publication sur un topic Event Grid
EVENTGRID_NAMESPACE_ENDPOINT=https://<namespace>.<region>.eventgrid.azure.net  # Requis pour les opérations sur namespace
AZURE_TOKEN_CREDENTIALS=prod # Requis uniquement si DefaultAzureCredential est utilisé en production

Authentification

import os
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.eventgrid import EventGridPublisherClient

# Dev local : DefaultAzureCredential. Production : définir AZURE_TOKEN_CREDENTIALS=prod ou AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Ou utiliser une credential spécifique directement en production :
# Voir https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

endpoint = "https://<topic-name>.<region>.eventgrid.azure.net/api/events"

client = EventGridPublisherClient(endpoint, credential)

Types d'événements

Publier des CloudEvents

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

client = EventGridPublisherClient(endpoint, DefaultAzureCredential())

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

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

Publier des EventGridEvents

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)

Propriétés des événements

Propriétés de CloudEvent

event = CloudEvent(
    type="MyApp.Events.ItemCreated",      # Requis : type d'événement
    source="/myapp/items",                 # Requis : source de l'événement
    data={"key": "value"},                 # Charge utile de l'événement
    subject="items/123",                   # Optionnel : sujet/chemin
    datacontenttype="application/json",   # Optionnel : type de contenu
    dataschema="https://schema.example",  # Optionnel : URL du schéma
    time=datetime.now(timezone.utc),      # Optionnel : horodatage
    extensions={"custom": "value"}         # Optionnel : attributs personnalisés
)

Propriétés d'EventGridEvent

event = EventGridEvent(
    subject="/myapp/items/123",            # Requis : sujet
    event_type="MyApp.ItemCreated",        # Requis : type d'événement
    data={"key": "value"},                 # Requis : charge utile de l'événement
    data_version="1.0",                    # Requis : version du schéma
    topic="/subscriptions/.../topics/...", # Optionnel : défini automatiquement
    event_time=datetime.now(timezone.utc)  # Optionnel : horodatage
)

Client asynchrone

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 (Namespaces Event Grid)

Pour les Namespaces Event Grid (livraison par pull) :

from azure.eventgrid.aio import EventGridPublisherClient

# Point de terminaison namespace (différent du topic personnalisé)
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
    )

Bonnes pratiques

1. Utiliser CloudEvents pour les nouvelles applications (standard industriel)
2. Regrouper les événements lors de la publication de plusieurs événements
3. Inclure des sujets significatifs pour le filtrage
4. Utiliser le client asynchrone pour les scénarios à haut débit
5. Gérer les tentatives — Event Grid dispose de mécanismes de retry intégrés
6. Définir des types d'événements appropriés pour le routage et le filtrage