biblioteca cliente de Azure Event Grid para Python: versión 4.16.0

  • Artículo

Azure Event Grid es un servicio de enrutamiento de eventos inteligente completamente administrado que permite el consumo de eventos uniforme mediante un modelo de publicación-suscripción.

Código | fuentePaquete (PyPI) | Paquete (Conda) | Documentación | de referencia de APIDocumentación | del productoMuestras | Changelog

Introducción

Requisitos previos

Instalar el paquete

Instale la biblioteca cliente de Azure Event Grid para Python con pip:

pip install azure-eventgrid
  • Se requiere un tema o dominio de Event Grid existente. Puede crear el recurso mediante Azure Portal o la CLI de Azure.

Si usa la CLI de Azure, reemplace <resource-group-name> y <resource-name> por sus propios nombres únicos.

Crear un tema de Event Grid

az eventgrid topic --create --location <location> --resource-group <resource-group-name> --name <resource-name>

Crear un dominio de Event Grid

az eventgrid domain --create --location <location> --resource-group <resource-group-name> --name <resource-name>

Autenticar el cliente

Para interactuar con el servicio Event Grid, deberá crear una instancia de un cliente. Se necesita un punto de conexión y una credencial para crear una instancia del objeto de cliente.

Uso de Azure Active Directory (AAD)

Azure Event Grid proporciona integración con Azure Active Directory (Azure AD) para la autenticación basada en identidades de las solicitudes. Con Azure AD, puede usar el control de acceso basado en rol (RBAC) para conceder acceso a los recursos de Azure Event Grid a usuarios, grupos o aplicaciones.

Para enviar eventos a un tema o dominio con , TokenCredentialla identidad autenticada debe tener asignado el rol "Remitente de datos de EventGrid".

Con el azure-identity paquete, puede autorizar sin problemas las solicitudes en entornos de desarrollo y producción. Para más información sobre Azure Active Directory, consulte el azure-identity archivo Léame.

Por ejemplo, puede usar DefaultAzureCredential para construir un cliente que se autenticará mediante Azure Active Directory:

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

default_az_credential = DefaultAzureCredential()
endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"]
client = EventGridPublisherClient(endpoint, default_az_credential)

Búsqueda del punto de conexión

Puede encontrar el punto de conexión del tema en el recurso tema de Event Grid en el Azure Portal. Esto tendrá un aspecto similar al siguiente: "https://<event-grid-topic-name>.<topic-location>.eventgrid.azure.net/api/events"

Creación del cliente con AzureKeyCredential

Para usar una clave de acceso como credential parámetro, pase la clave como una cadena a una instancia de AzureKeyCredential.

Nota: La clave de acceso se puede encontrar en Azure Portal en el menú "Claves de acceso" del recurso tema de Event Grid. También se pueden obtener a través de la CLI de Azure o la azure-mgmt-eventgrid biblioteca. Puede encontrar una guía para obtener las claves de acceso aquí.

import os
from azure.eventgrid import EventGridPublisherClient
from azure.core.credentials import AzureKeyCredential

topic_key = os.environ["EVENTGRID_TOPIC_KEY"]
endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"]

credential_key = AzureKeyCredential(topic_key)
client = EventGridPublisherClient(endpoint, credential_key)

Nota: Un cliente también se puede autenticar a través de la firma SAS, mediante .AzureSasCredential Un ejemplo que muestra esto está disponible aquí (async_version).

Nota: El generate_sas método se puede usar para generar una firma de acceso compartido. Aquí se puede ver un ejemplo que muestra esto.

Conceptos clave

Tema

Un tema es un canal dentro del servicio EventGrid para enviar eventos. El esquema de eventos que acepta un tema se decide en el momento de la creación del tema. Si los eventos de un tipo de esquema se envían a un tema que requiere un tipo de esquema diferente, se producirán errores.

Dominio

Un dominio de eventos es una herramienta de administración para un gran número de temas de Event Grid relacionados con la misma aplicación. Le permiten publicar eventos en miles de temas. Los dominios también le proporcionan autorización y control de autenticación sobre cada tema. Para obtener más información, visite Introducción al dominio de eventos.

Cuando se crea un dominio de eventos, se pone a su disposición un punto de conexión de publicación para este dominio. Este proceso es similar a la creación de un tema de Event Grid. La única diferencia es que, al publicar en un dominio, debe especificar el tema dentro del dominio al que desea que se entregue el evento.

Esquemas de eventos

Un evento es la menor cantidad de información que describe completamente algo que sucedió en el sistema. Cuando se crea un tema o dominio personalizado, debe especificar el esquema que se usará al publicar eventos.

Event Grid admite varios esquemas para codificar eventos.

Esquema de Event Grid

Aunque puede configurar el tema para usar un esquema personalizado, es más común usar el esquema de Event Grid ya definido. Consulte las especificaciones y los requisitos aquí.

Esquema de CloudEvents v1.0

Otra opción es usar el esquema CloudEvents v1.0. CloudEvents es un proyecto de Cloud Native Computing Foundation que genera una especificación para describir los datos de eventos de una manera común. Puede encontrar el resumen del servicio de CloudEvents aquí.

EventGridPublisherClient

EventGridPublisherClient proporciona operaciones para enviar datos de eventos a un nombre de host de tema especificado durante la inicialización del cliente.

Independientemente del esquema que el tema o dominio esté configurado para su uso, EventGridPublisherClient se usará para publicar eventos en él. Use los eventos de publicación de send métodos.

Se permiten enviar los siguientes formatos de eventos:

  • Una lista o una sola instancia de EventGridEvents fuertemente tipadas.

  • Representación dict de un objeto EventGridEvent serializado.

  • Una lista o una sola instancia de CloudEvents fuertemente tipadas.

  • Representación dict de un objeto CloudEvent serializado.

  • Representación dict de cualquier esquema personalizado.

Consulte los ejemplos para obtener ejemplos detallados.

Nota: Es importante saber si el tema admite CloudEvents o EventGridEvents antes de publicarlo. Si envía a un tema que no admite el esquema del evento que está enviando, send() producirá una excepción.

Temas del sistema

Un tema del sistema de Event Grid representa uno o varios eventos publicados por servicios de Azure, como Azure Storage o Azure Event Hubs. Por ejemplo, un tema del sistema puede representar todos los eventos de blob o solo los eventos de creación de blobs y eliminación de blobs publicados para una cuenta de almacenamiento específica.

Los nombres de los distintos tipos de eventos para los eventos del sistema publicados en Azure Event Grid están disponibles en azure.eventgrid.SystemEventNames. Para obtener una lista completa de temas reconocibles del sistema, visite Temas del sistema.

Para más información sobre los conceptos clave de Event Grid, consulte Conceptos en Azure Event Grid.

Event Grid en Kubernetes con Azure Arc

Event Grid en Kubernetes con Azure Arc es una oferta que le permite ejecutar Event Grid en su propio clúster de Kubernetes. Esta funcionalidad se habilita mediante el uso de Kubernetes habilitado para Azure Arc. Mediante Kubernetes habilitado para Azure Arc, se conecta a Azure un clúster de Kubernetes compatible. Una vez conectado, puede instalar Event Grid en él. Obtenga más información al respecto aquí.

Compatibilidad con eventos en la nube de CNCF

A partir de v4.7.0, este paquete también admite la publicación de un evento en la nube CNCF desde https://pypi.org/project/cloudevents/. Podría pasar un objeto CloudEvent de esta biblioteca a la send API.


from cloudevents.http import CloudEvent

event = CloudEvent(...)

client.send(event)

Ejemplos

En las secciones siguientes se proporcionan varios fragmentos de código que abarcan algunas de las tareas más comunes de Event Grid, entre las que se incluyen:

Envío de un evento de Event Grid

En este ejemplo se publica un evento de Event Grid.

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

key = os.environ["EG_ACCESS_KEY"]
endpoint = os.environ["EG_TOPIC_HOSTNAME"]

event = EventGridEvent(
    data={"team": "azure-sdk"},
    subject="Door1",
    event_type="Azure.Sdk.Demo",
    data_version="2.0"
)

credential = AzureKeyCredential(key)
client = EventGridPublisherClient(endpoint, credential)

client.send(event)

Envío de un evento en la nube

En este ejemplo se publica un evento en la nube.

import os
from azure.core.credentials import AzureKeyCredential
from azure.core.messaging import CloudEvent
from azure.eventgrid import EventGridPublisherClient

key = os.environ["CLOUD_ACCESS_KEY"]
endpoint = os.environ["CLOUD_TOPIC_HOSTNAME"]

event = CloudEvent(
    type="Azure.Sdk.Sample",
    source="https://egsample.dev/sampleevent",
    data={"team": "azure-sdk"}
)

credential = AzureKeyCredential(key)
client = EventGridPublisherClient(endpoint, credential)

client.send(event)

Envío de varios eventos

Es posible enviar eventos como un lote al enviar varios eventos a un tema o un dominio. En este ejemplo se envía una lista de CloudEvents mediante el método send.

ADVERTENCIA: Al enviar una lista de varios eventos a la vez, la iteración y el envío de cada evento no dará como resultado un rendimiento óptimo. Para obtener el mejor rendimiento, se recomienda enviar una lista de eventos.

import os
from azure.core.credentials import AzureKeyCredential
from azure.core.messaging import CloudEvent
from azure.eventgrid import EventGridPublisherClient

key = os.environ["CLOUD_ACCESS_KEY"]
endpoint = os.environ["CLOUD_TOPIC_HOSTNAME"]

event0 = CloudEvent(
    type="Azure.Sdk.Sample",
    source="https://egsample.dev/sampleevent",
    data={"team": "azure-sdk"}
)
event1 = CloudEvent(
    type="Azure.Sdk.Sample",
    source="https://egsample.dev/sampleevent",
    data={"team2": "azure-eventgrid"}
)

events = [event0, event1]

credential = AzureKeyCredential(key)
client = EventGridPublisherClient(endpoint, credential)

client.send(events)

Envío de eventos como diccionarios

También se puede usar una representación dict de los modelos serializados respectivos para publicar Eventos en la nube o EventGridEvent(s) aparte de los objetos fuertemente tipados.

Use una representación de tipo dict para enviar a un tema con un esquema personalizado, como se muestra a continuación.

import os
import uuid
import datetime as dt
from msrest.serialization import UTC
from azure.core.credentials import AzureKeyCredential
from azure.eventgrid import EventGridPublisherClient

key = os.environ["CUSTOM_SCHEMA_ACCESS_KEY"]
endpoint = os.environ["CUSTOM_SCHEMA_TOPIC_HOSTNAME"]

event = custom_schema_event = {
    "customSubject": "sample",
    "customEventType": "sample.event",
    "customDataVersion": "2.0",
    "customId": uuid.uuid4(),
    "customEventTime": dt.datetime.now(UTC()).isoformat(),
    "customData": "sample data"
    }

credential = AzureKeyCredential(key)
client = EventGridPublisherClient(endpoint, credential)

client.send(event)

Consumir desde la cola de almacenamiento

En este ejemplo se consume un mensaje recibido de la cola de almacenamiento y se deserializa en un objeto CloudEvent.

from azure.core.messaging import CloudEvent
from azure.storage.queue import QueueServiceClient, BinaryBase64DecodePolicy
import os
import json

# all types of CloudEvents below produce same DeserializedEvent
connection_str = os.environ['STORAGE_QUEUE_CONN_STR']
queue_name = os.environ['STORAGE_QUEUE_NAME']

with QueueServiceClient.from_connection_string(connection_str) as qsc:
    payload =  qsc.get_queue_client(
        queue=queue_name,
        message_decode_policy=BinaryBase64DecodePolicy()
        ).peek_messages()

    ## deserialize payload into a list of typed Events
    events = [CloudEvent.from_dict(json.loads(msg.content)) for msg in payload]

Consumir desde servicebus

En este ejemplo se consume un mensaje de carga recibido de ServiceBus y se deserializa en un objeto EventGridEvent.

from azure.eventgrid import EventGridEvent
from azure.servicebus import ServiceBusClient
import os
import json

# all types of EventGridEvents below produce same DeserializedEvent
connection_str = os.environ['SERVICE_BUS_CONN_STR']
queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']

with ServiceBusClient.from_connection_string(connection_str) as sb_client:
    payload =  sb_client.get_queue_receiver(queue_name).receive_messages()

    ## deserialize payload into a list of typed Events
    events = [EventGridEvent.from_dict(json.loads(next(msg.body).decode('utf-8'))) for msg in payload]

Seguimiento distribuido con EventGrid

Puede usar OpenTelemetry para Python como de costumbre con EventGrid, ya que es compatible con la integración del seguimiento principal de Azure.

Este es un ejemplo del uso de OpenTelemetry para realizar un seguimiento del envío de un objeto CloudEvent.

En primer lugar, establezca OpenTelemetry como complemento de seguimiento habilitado para EventGrid.

from azure.core.settings import settings
from azure.core.tracing.ext.opentelemetry_span import OpenTelemetrySpan

settings.tracing_implementation = OpenTelemetrySpan

Uso de telemetría abierto normalmente desde aquí. Consulte OpenTelemetry para obtener más información. En este ejemplo se usa un exportador de consola simple para exportar los seguimientos. Cualquier exportador se puede usar aquí, como azure-monitor-opentelemetry-exporter, jaeger, etc zipkin .

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter
from opentelemetry.sdk.trace.export import SimpleSpanProcessor  # this requires opentelemetry >= 1.0.0

# Simple console exporter
exporter = ConsoleSpanExporter()

trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)
trace.get_tracer_provider().add_span_processor(
    SimpleSpanProcessor(exporter)
)

tracer Una vez establecido y exporter , siga el ejemplo siguiente para empezar a recopilar seguimientos mientras se usa el send método de EventGridPublisherClient para enviar un objeto CloudEvent.

import os
from azure.eventgrid import EventGridPublisherClient
from azure.core.messaging import CloudEvent
from azure.core.credentials import AzureKeyCredential

hostname = os.environ['CLOUD_TOPIC_HOSTNAME']
key = AzureKeyCredential(os.environ['CLOUD_ACCESS_KEY'])
cloud_event = CloudEvent(
    source = 'demo',
    type = 'sdk.demo',
    data = {'test': 'hello'},
)
with tracer.start_as_current_span(name="MyApplication"):
    client = EventGridPublisherClient(hostname, key)
    client.send(cloud_event)

Solución de problemas

  • Habilite el azure.eventgrid registrador para recopilar seguimientos de la biblioteca.

General

La biblioteca cliente de Event Grid generará excepciones definidas en Azure Core.

Registro

Esta biblioteca usa la biblioteca de registro estándar para el registro. La información básica sobre las sesiones HTTP (direcciones URL, encabezados, etc.) se registra en el nivel INFO.

Configuración opcional

Los argumentos de palabra clave opcionales se pueden pasar en el nivel de cliente y por operación. En la documentación de referencia de azure-core se describen las configuraciones disponibles para reintentos, registro, protocolos de transporte, etc.

Pasos siguientes

En la sección siguiente se proporcionan varios fragmentos de código que ilustran patrones comunes que se usan en la API de Python de Event Grid.

Más código de ejemplo

Estos ejemplos de código muestran operaciones de escenario de campeón comunes con la biblioteca cliente de Azure Event Grid.

En los ejemplos siguientes se tratan la publicación y el consumo dict de representaciones de EventGridEvents y CloudEvents.

Puede encontrar más ejemplos aquí.

  • Puede ver más ejemplos relacionados con el escenario de envío aquí.
  • Para ver más ejemplos relacionados con el consumo de una carga de diferentes servicios de mensajería como un objeto con tipo, visite Consumir ejemplos.

Documentación adicional

Para obtener documentación más amplia sobre Azure Event Grid, consulte la documentación de Event Grid sobre docs.microsoft.com.

Contribuciones

Este proyecto agradece las contribuciones y sugerencias. La mayoría de las contribuciones requieren que acepte un Contrato de licencia para el colaborador (CLA) que declara que tiene el derecho a concedernos y nos concede los derechos para usar su contribución. Para más información, visite cla.microsoft.com.

Cuando se envía una solicitud de incorporación de cambios, un bot de CLA determinará de forma automática si tiene que aportar un CLA y completar la PR adecuadamente (por ejemplo, la etiqueta, el comentario). Solo siga las instrucciones que le dará el bot. Solo será necesario que lo haga una vez en todos los repositorios con nuestro CLA.

Este proyecto ha adoptado el Código de conducta de Microsoft Open Source. Para más información, consulte las preguntas más frecuentes del código de conducta o póngase en contacto con opencode@microsoft.com si tiene cualquier otra pregunta o comentario.