Share via

Azure Event Grid biblioteca de cliente para Python - versão 4.16.0

  • Artigo

O Azure Event Grid é um serviço inteligente de encaminhamento de eventos integralmente gerido que permite um consumo de eventos uniforme através de um modelo de publicação-subscrição.

Código fonte | Pacote (PyPI) | Pacote (Conda) | Documentação | de referência da APIDocumentação do | produtoExemplos | Registo de alterações

Introdução

Pré-requisitos

Instalar o pacote

Instale a biblioteca de cliente Azure Event Grid para Python com pip:

pip install azure-eventgrid

Se utilizar a CLI do Azure, substitua <resource-group-name> e <resource-name> pelos seus próprios nomes exclusivos.

Criar um Tópico do Event Grid

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

Criar um Domínio do Event Grid

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

Autenticar o cliente

Para interagir com o serviço Event Grid, terá de criar uma instância de um cliente. É necessário um ponto final e uma credencial para instanciar o objeto de cliente.

Utilizar o Azure Active Directory (AAD)

Azure Event Grid fornece integração com o Azure Active Directory (Azure AD) para autenticação baseada em identidade de pedidos. Com Azure AD, pode utilizar o controlo de acesso baseado em funções (RBAC) para conceder acesso aos seus recursos Azure Event Grid a utilizadores, grupos ou aplicações.

Para enviar eventos para um tópico ou domínio com um TokenCredential, a identidade autenticada deve ter a função "EventGrid Data Sender" atribuída.

Com o azure-identity pacote, pode autorizar pedidos de forma totalmente integrada em ambientes de desenvolvimento e produção. Para saber mais sobre o Azure Active Directory, veja o azure-identity README.

Por exemplo, pode utilizar DefaultAzureCredential para construir um cliente que se autenticará com o 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)

Procurar o ponto final

Pode encontrar o ponto final do tópico no recurso tópico do Event Grid no portal do Azure. Terá o seguinte aspeto: "https://<event-grid-topic-name>.<topic-location>.eventgrid.azure.net/api/events"

Criar o cliente com o AzureKeyCredential

Para utilizar uma chave de Acesso como parâmetro credential , transmita a chave como uma cadeia para uma instância do AzureKeyCredential.

Nota: A Chave de Acesso pode ser encontrada no portal do Azure no menu "Chaves de Acesso" do recurso do Tópico do Event Grid. Também podem ser obtidos através da CLI do Azure ou da azure-mgmt-eventgrid biblioteca. Pode encontrar um guia para obter chaves de acesso aqui.

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: Um cliente também pode ser autenticado através da assinatura SAS, com o AzureSasCredential. Está disponível um exemplo que demonstra isto aqui (async_version).

Nota: O generate_sas método pode ser utilizado para gerar uma assinatura de acesso partilhado. Pode ver um exemplo que demonstra isto aqui.

Conceitos-chave

Tópico

Um tópico é um canal no serviço EventGrid para enviar eventos. O esquema de evento que um tópico aceita é decidido no momento da criação do tópico. Se os eventos de um tipo de esquema forem enviados para um tópico que requer um tipo de esquema diferente, serão gerados erros.

Domínio

Um domínio de eventos é uma ferramenta de gestão para um grande número de tópicos do Event Grid relacionados com a mesma aplicação. Permitem-lhe publicar eventos em milhares de tópicos. Os domínios também lhe dão autorização e controlo de autenticação sobre cada tópico. Para obter mais informações, visite Descrição geral do domínio de eventos.

Quando cria um domínio de evento, é disponibilizado um ponto final de publicação para este domínio. Este processo é semelhante à criação de um Tópico do Event Grid. A única diferença é que, ao publicar num domínio, tem de especificar o tópico no domínio ao qual pretende que o evento seja entregue.

Esquemas de eventos

Um evento é a menor quantidade de informações que descreve totalmente algo que aconteceu no sistema. Quando é criado um tópico ou domínio personalizado, tem de especificar o esquema que será utilizado ao publicar eventos.

O Event Grid suporta vários esquemas para eventos de codificação.

Esquema do Event Grid

Embora possa configurar o tópico para utilizar um esquema personalizado, é mais comum utilizar o esquema do Event Grid já definido. Veja as especificações e os requisitos aqui.

Esquema CloudEvents v1.0

Outra opção é utilizar o esquema CloudEvents v1.0. O CloudEvents é um projeto do Cloud Native Computing Foundation que produz uma especificação para descrever os dados de eventos de uma forma comum. O resumo do serviço do CloudEvents pode ser encontrado aqui.

EventGridPublisherClient

EventGridPublisherClient fornece operações para enviar dados de eventos para um nome de anfitrião de tópico especificado durante a inicialização do cliente.

Independentemente do esquema que o seu tópico ou domínio está configurado para utilizar, EventGridPublisherClient será utilizado para publicar eventos no mesmo. Utilize os eventos de publicação do send método.

Os seguintes formatos de eventos podem ser enviados:

  • Uma lista ou uma única instância de EventGridEvents fortemente escritos.

  • Uma representação dict de um objeto EventGridEvent serializado.

  • Uma lista ou uma única instância de CloudEvents fortemente escritos.

  • Uma representação dict de um objeto CloudEvent serializado.

  • Uma representação de ditado de qualquer Esquema Personalizado.

Veja os exemplos para obter exemplos detalhados.

Nota: É importante saber se o seu tópico suporta CloudEvents ou EventGridEvents antes de publicar. Se enviar para um tópico que não suporta o esquema do evento que está a enviar, enviar() apresentará uma exceção.

Tópicos do Sistema

Um tópico de sistema no Event Grid representa um ou mais eventos publicados pelos serviços do Azure, como o Armazenamento do Azure ou Hubs de Eventos do Azure. Por exemplo, um tópico de sistema pode representar todos os eventos de blobs ou apenas eventos de criação e eliminação de blobs publicados para uma conta de armazenamento específica.

Os nomes dos vários tipos de eventos para os eventos de sistema publicados no Azure Event Grid estão disponíveis no azure.eventgrid.SystemEventNames. Para obter uma lista completa de tópicos de sistema reconhecíveis, visite Tópicos do Sistema.

Para obter mais informações sobre os principais conceitos no Event Grid, veja Conceitos no Azure Event Grid.

Event Grid no Kubernetes com o Azure Arc

O Event Grid no Kubernetes com o Azure Arc é uma oferta que lhe permite executar o Event Grid no seu próprio cluster do Kubernetes. Esta capacidade é ativada pela utilização do Kubernetes compatível com o Azure Arc. Através do Kubernetes compatível com o Azure Arc, um cluster do Kubernetes suportado liga-se ao Azure. Assim que estiver ligado, poderá instalar o Event Grid no mesmo. Saiba mais sobre o assunto aqui.

Suporte para Eventos na Cloud CNCF

A partir da v4.7.0, este pacote também suporta a publicação de um evento na cloud CNCF a partir de https://pypi.org/project/cloudevents/. Poderia transmitir um objeto CloudEvent desta biblioteca para a send API.


from cloudevents.http import CloudEvent

event = CloudEvent(...)

client.send(event)

Exemplos

As secções seguintes fornecem vários fragmentos de código que abrangem algumas das tarefas mais comuns do Event Grid, incluindo:

Enviar um Evento do Event Grid

Este exemplo publica um evento do 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)

Enviar um Evento na Cloud

Este exemplo publica um evento da Cloud.

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)

Enviar Vários eventos

É possível enviar eventos como um lote ao enviar vários eventos para um tópico ou domínio. Este exemplo envia uma lista de CloudEvents com o método de envio.

AVISO: Ao enviar uma lista de vários eventos de uma só vez, iterar e enviar cada evento não resultará num desempenho ideal. Para obter o melhor desempenho, é altamente recomendado enviar uma 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)

Enviar eventos como dicionários

Uma representação de ditado dos respetivos modelos serializados também pode ser utilizada para publicar CloudEvent(s) ou EventGridEvent(s) para além dos objetos fortemente escritos.

Utilize uma representação semelhante a um ditado para enviar para um tópico com esquema personalizado, conforme mostrado abaixo.

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 a partir da fila de armazenamento

Este exemplo consome uma mensagem recebida da fila de armazenamento e desseleciona-a para um 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 a partir do servicebus

Este exemplo consome uma mensagem de payload recebida do ServiceBus e aninha-a para um 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]

Rastreio Distribuído com EventGrid

Pode utilizar o OpenTelemetry para Python como habitualmente com o EventGrid, uma vez que é compatível com a integração de rastreio do azure-core.

Eis um exemplo de utilização do OpenTelemetry para rastrear o envio de um CloudEvent.

Primeiro, defina OpenTelemetry como plug-in de rastreio ativado para EventGrid.

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

settings.tracing_implementation = OpenTelemetrySpan

Utilização de telemetria aberta regular a partir daqui. Consulte OpenTelemetry para obter detalhes. Este exemplo utiliza um exportador de consola simples para exportar os rastreios. Qualquer exportador pode ser utilizado aqui, incluindo azure-monitor-opentelemetry-exporter, jaeger, zipkin etc.

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)
)

Assim que o tracer e exporter estiver definido, siga o exemplo abaixo para começar a recolher rastreios ao utilizar o send método do EventGridPublisherClient para enviar um 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)

Resolução de problemas

  • Ative azure.eventgrid o logger para recolher rastreios da biblioteca.

Geral

A biblioteca de cliente do Event Grid irá gerar exceções definidas no Azure Core.

Registo

Esta biblioteca utiliza a biblioteca de registos padrão para registos. As informações básicas sobre as sessões HTTP (URLs, cabeçalhos, etc.) são registadas ao nível da INFORMAÇÃO.

Configuração opcional

Os argumentos de palavra-chave opcionais podem ser transmitidos ao nível do cliente e por operação. A documentação de referência do azure-core descreve as configurações disponíveis para repetições, registos, protocolos de transporte e muito mais.

Passos seguintes

A secção seguinte fornece vários fragmentos de código que ilustram padrões comuns utilizados na API python do Event Grid.

Mais código de exemplo

Estes exemplos de código mostram operações de cenário de campeão comuns com a biblioteca de cliente Azure Event Grid.

Os exemplos seguintes abrangem a publicação e o consumo dict de representações de EventGridEvents e CloudEvents.

Pode encontrar mais exemplos aqui.

  • Pode ver mais exemplos relacionados com o cenário de envio aqui.
  • Para ver mais exemplos relacionados com o consumo de um payload de diferentes serviços de mensagens como um objeto escrito, visite Consumir Amostras

Documentação adicional

Para obter documentação mais extensa sobre Azure Event Grid, consulte a documentação do Event Grid sobre docs.microsoft.com.

Contribuir

Agradecemos todas as contribuições e sugestões para este projeto. A maioria das contribuições requerem que celebre um Contrato de Licença de Contribuição (CLA) no qual se declare que tem o direito de conceder e que, na verdade, concede-nos os direitos para utilizar a sua contribuição. Para obter detalhes, visite cla.microsoft.com.

Quando submete um pedido Pull, um bot do CLA determina automaticamente se tem de fornecer um CLA e decorar o PR de forma adequada (por exemplo, etiqueta, comentário). Só tem de seguir as instruções fornecidas pelo bot. Apenas terá de fazer isto uma vez em todos os repositórios com o nosso CLA.

Este projeto adotou o Microsoft Open Source Code of Conduct (Código de Conduta do Microsoft Open Source). Para obter mais informações, consulte as FAQ do Código de Conduta ou o contacto opencode@microsoft.com com quaisquer perguntas ou comentários adicionais.