Enviar ou receber eventos dos Hubs de Eventos usando o Python

Artigo
02/07/2024

Este início rápido mostra como enviar e receber eventos de um hub de eventos usando o pacote Python azure-eventhub.

Pré-requisitos

Se você estiver conhecendo agora os Hubs de Eventos do Azure, confira Visão geral dos Hubs de Eventos antes de prosseguir com este início rápido.

Para concluir este início rápido, você precisará dos seguintes pré-requisitos:

Assinatura do Microsoft Azure. Para usar os serviços do Azure, incluindo os Hubs de Eventos do Azure, você precisa ter uma assinatura. Caso não tenha uma conta do Azure existente, inscreva-se para uma avaliação gratuita.
Python 3.8 ou posterior com PIP instalado e atualizado.
Visual Studio Code (recomendado) ou qualquer outro IDE (ambiente de desenvolvimento integrado).
Criar um namespace de Hubs de Eventos e um hub de eventos. A primeira etapa é usar o portal do Azure para criar um namespace dos Hubs de Eventos e obter as credenciais de gerenciamento que seu aplicativo precisa para se comunicar com o hub de eventos. Para criar um namespace e um hub de eventos, siga o procedimento nesse artigo.

Instalar os pacotes para enviar eventos

Para instalar os pacotes do Python para Hubs de Eventos, abra um prompt de comando que tenha Python no caminho. Altere o diretório para a pasta em que você deseja manter os exemplos.

Sem senha (recomendado)
Cadeia de Conexão

pip install azure-eventhub
pip install azure-identity
pip install aiohttp

pip install azure-eventhub

Autenticar o aplicativo no Azure

Este guia de início rápido mostra duas maneiras de se conectar aos Hubs de Eventos do Azure: sem senha e cadeia de conexão. A primeira opção mostra como usar sua entidade de segurança no Microsoft Entra ID e no RBAC (controle de acesso baseado em função) para se conectar a um namespace dos Hubs de Eventos. Você não precisa se preocupar em ter cadeias de conexão embutidas no código, em um arquivo de configuração ou em um armazenamento seguro, como o Azure Key Vault. A segunda opção mostra como usar uma cadeia de conexão para se conectar a um namespace dos Hubs de Eventos. Se você é novo no Azure, talvez ache a opção de cadeia de conexão mais fácil de ser seguida. É recomendável usar a opção sem senha em aplicativos do mundo real e ambientes de produção. Para obter mais informações, confira Autenticação e autorização. Você também pode ler mais sobre a autenticação sem senha na página de visão geral.

Sem senha (recomendado)
Cadeia de Conexão

Atribuir funções ao usuário do Microsoft Entra

No caso de desenvolvimento local, verifique se a conta de usuário que se conecta aos Hubs de Eventos do Azure tem as permissões corretas. Você precisará da função Proprietário de Dados dos Hubs de Eventos do Azure para enviar e receber mensagens. Para atribuir essa função a si mesmo, você precisará da função Administrador de acesso do usuário ou de outra função que inclua a ação Microsoft.Authorization/roleAssignments/write. É possível atribuir funções RBAC do Azure a um usuário usando o portal do Azure, a CLI do Azure ou o Azure PowerShell. Saiba mais sobre os escopos disponíveis para atribuições de função na página de visão geral do escopo.

O exemplo a seguir atribui a função Azure Event Hubs Data Owner à sua conta de usuário, que fornece acesso completo aos recursos dos Hubs de Eventos do Azure. Em um cenário real, siga o Princípio do Privilégio Mínimo para dar aos usuários apenas as permissões mínimas necessárias para um ambiente de produção mais seguro.

Funções interna do Azure par Hubs de Eventos do Azure

Para os Hubs de Eventos do Azure, o gerenciamento de namespaces e de todos os recursos relacionados por meio do portal do Azure e da API de gerenciamento de recursos do Azure já está protegido pelo modelo Azure RBAC. O Azure fornece as funções internas do Azure abaixo para autorizar o acesso a um namespace dos Hubs de Eventos:

Proprietário de Dados dos Hubs de Eventos do Azure: permite o acesso a dados no namespace dos Hubs de Eventos e às respectivas entidades (filas, tópicos, assinaturas e filtros)
Remetente de Dados dos Hubs de Eventos do Azure: use essa função para fornecer acesso de remetente ao namespace dos Hubs de Eventos e às respectivas entidades.
Destinatário de Dados dos Hubs de Eventos do Azure: use essa função para fornecer acesso de remetente ao namespace dos Hubs de Eventos e às respectivas entidades.

Se você quiser criar uma função personalizada, confira Direitos exigidos para operações dos Hubs de Eventos.

Importante

Na maioria dos casos, levará um ou dois minutos para a atribuição de função se propagar no Azure. Em casos raros, poderá levar até oito minutos. Se você receber erros de autenticação ao executar o código pela primeira vez, aguarde alguns instantes e tente novamente.

No portal do Azure, localize o namespace dos Hubs de Eventos usando a barra de pesquisa principal ou a navegação à esquerda.
Na página de visão geral, selecione Controle de acesso (IAM) no menu à esquerda.
Na página Controle de acesso (IAM), selecione a guia Atribuições de função.
Selecione + Adicionar no menu superior e, em seguida, Adicionar atribuição de função no menu suspenso resultante.
Use a caixa de pesquisa para filtrar os resultados para a função desejada. Neste exemplo, pesquise Azure Event Hubs Data Owner e selecione o resultado correspondente. Em seguida, escolha Avançar.
Em Atribuir acesso a, selecione Usuário, grupo ou entidade de serviço e, em seguida, selecione + Selecionar membros.
No diálogo, pesquise seu nome de usuário do Microsoft Entra (geralmente seu endereço de email user@domain) e escolha Selecionar na parte inferior do diálogo.
Selecione Revisar + atribuir para ir para a página final e, em seguida, Revisar + atribuir novamente para concluir o processo.

Para atribuir uma função no nível do recurso usando a CLI do Azure, primeiro você deve recuperar a ID do recurso usando o comando az eventhubs namespace show. É possível filtrar as propriedades de saída usando o parâmetro --query.

az eventhubs namespace show -g '<your-event-hub-resource-group>' -n '<your-event-hub-name> --query id

Copie a saída Id do comando anterior. Em seguida, você pode atribuir funções usando o comando az role da CLI do Azure.

az role assignment create --assignee "<user@domain>" \
--role "Azure Event Hubs Data Owner" \
--scope "<your-resource-id>"

Para atribuir uma função no nível do recurso usando o Azure PowerShell, primeiro você precisa recuperar a ID do recurso usando o comando Get-AzResource.

Get-AzResource -ResourceGroupName "<your-event-hub-resource-group>" -Name "<your-event-hub-name>"

Copie o valor Id da saída do comando anterior. Em seguida, é possível atribuir funções usando o comando New-AzRoleAssignment no PowerShell.

New-AzRoleAssignment -SignInName <user@domain> `
-RoleDefinitionName "Azure Event Hubs Data Owner" `
-Scope <yourResourceId>

Enviar eventos

Nesta seção, crie um script Python para enviar eventos para o hub de eventos que você criou anteriormente.

Abra seu editor favorito do Python, como o Visual Studio Code.
Crie um script chamado send.py. Esse script envia um lote de eventos para o hub de eventos que você criou anteriormente.

Cole o seguinte código em sender.py:

Sem senha (recomendado)
Cadeia de Conexão

No código, use valores reais para substituir os seguintes espaços reservados:

EVENT_HUB_FULLY_QUALIFIED_NAMESPACE
EVENT_HUB_NAME

import asyncio

from azure.eventhub import EventData
from azure.eventhub.aio import EventHubProducerClient
from azure.identity.aio import DefaultAzureCredential

EVENT_HUB_FULLY_QUALIFIED_NAMESPACE = "EVENT_HUB_FULLY_QUALIFIED_NAMESPACE"
EVENT_HUB_NAME = "EVENT_HUB_NAME"

credential = DefaultAzureCredential()

async def run():
    # Create a producer client to send messages to the event hub.
    # Specify a credential that has correct role assigned to access
    # event hubs namespace and the event hub name.
    producer = EventHubProducerClient(
        fully_qualified_namespace=EVENT_HUB_FULLY_QUALIFIED_NAMESPACE,
        eventhub_name=EVENT_HUB_NAME,
        credential=credential,
    )
    async with producer:
        # Create a batch.
        event_data_batch = await producer.create_batch()

        # Add events to the batch.
        event_data_batch.add(EventData("First event "))
        event_data_batch.add(EventData("Second event"))
        event_data_batch.add(EventData("Third event"))

        # Send the batch of events to the event hub.
        await producer.send_batch(event_data_batch)

        # Close credential when no longer needed.
        await credential.close()

asyncio.run(run())

No código, use valores reais para substituir os seguintes espaços reservados:

EVENT_HUB_CONNECTION_STR
EVENT_HUB_NAME

Observação

Verifique se EVENT_HUB_NAME é o nome do hub de eventos e não o namespace dos Hubs de Eventos. Se esse valor estiver incorreto, você receberá o código de erro: CBS Token authentication failed..

import asyncio

from azure.eventhub import EventData
from azure.eventhub.aio import EventHubProducerClient

EVENT_HUB_CONNECTION_STR = "EVENT_HUB_CONNECTION_STR"
EVENT_HUB_NAME = "EVENT_HUB_NAME"

async def run():
    # Create a producer client to send messages to the event hub.
    # Specify a connection string to your event hubs namespace and
    # the event hub name.
    producer = EventHubProducerClient.from_connection_string(
        conn_str=EVENT_HUB_CONNECTION_STR, eventhub_name=EVENT_HUB_NAME
    )
    async with producer:
        # Create a batch.
        event_data_batch = await producer.create_batch()

        # Add events to the batch.
        event_data_batch.add(EventData("First event "))
        event_data_batch.add(EventData("Second event"))
        event_data_batch.add(EventData("Third event"))

        # Send the batch of events to the event hub.
        await producer.send_batch(event_data_batch)

asyncio.run(run())

Observação

Para ver exemplos de outras opções para enviar eventos ao Hub de Eventos de maneira assíncrona usando uma cadeia de conexão, confira a página send_async.py do GitHub. Os padrões mostrados também são aplicáveis ao envio de eventos sem senha.

Receber eventos

Este início rápido usa o Armazenamento de Blobs do Azure como um repositório de pontos de verificação. O repositório de pontos de verificação é usado para persistir pontos de verificação (ou seja, as últimas posições de leitura).

Siga estas recomendações ao usar Armazenamento de Blobs do Azure como um repositório de ponto de verificação:

Use um contêiner separado para cada grupo de consumidores. Você pode usar a mesma conta de armazenamento, mas usar um contêiner por cada grupo.
Não use o contêiner para mais nada e não use a conta de armazenamento para mais nada.
A conta de armazenamento deve estar na mesma região em que o aplicativo implantado está localizado. Se o aplicativo for local, tente escolher a região mais próxima possível.

Na página Conta de armazenamento do portal do Azure, na seção serviço Blob, verifique se as seguintes configurações estão desabilitadas.

Namespace hierárquico
Exclusão temporária de blobs
Controle de versão

Criar uma conta de armazenamento e um contêiner de blobs do Azure

Crie uma conta de armazenamento do Azure e um contêiner de blobs nela executando as seguintes etapas:

Crie uma conta de Armazenamento do Azure
Criar um contêiner de blob.
Autenticar no contêiner de blob.

Registre a cadeia de conexão e o nome do contêiner para usá-los posteriormente no código de recebimento.

Sem senha (recomendado)
Cadeia de Conexão

Ao desenvolver localmente, verifique se a conta de usuário que está acessando os dados de blob tem as permissões corretas. Você precisará do Colaborador de Dados do Blob de Armazenamento para ler e gravar os dados de blob. Para atribuir essa função a si mesmo, você precisará receber a atribuição da função Administrador de Acesso do Usuário ou de outra função que inclua a ação Microsoft.Authorization/roleAssignments/write. É possível atribuir funções RBAC do Azure a um usuário usando o portal do Azure, a CLI do Azure ou o Azure PowerShell. Você pode saber mais sobre os escopos disponíveis para atribuições de função na página de visão geral do escopo.

Nesse cenário, você atribuirá permissões à sua conta de usuário, no escopo da conta de armazenamento, para seguir o Princípio do Privilégio Mínimo. Essa prática fornece aos usuários apenas as permissões mínimas necessárias e cria ambientes de produção mais seguros.

O exemplo a seguir atribuirá a função de Colaborador de Dados do Blob de Armazenamento à sua conta de usuário, que fornece acesso de leitura e gravação aos dados de blob na sua conta de armazenamento.

Importante

Na maioria dos casos, levará um ou dois minutos para a atribuição de função se propagar no Azure, mas em casos raros pode levar até oito minutos. Se você receber erros de autenticação ao executar o código pela primeira vez, aguarde alguns instantes e tente novamente.

No portal do Azure, localize sua conta de armazenamento usando a barra de pesquisa principal ou a navegação à esquerda.
Na página de visão geral da conta de armazenamento, selecione Controle de acesso (IAM) no menu à esquerda.
Na página Controle de acesso (IAM), selecione a guia Atribuições de função.
Selecione + Adicionar no menu superior e, em seguida, Adicionar atribuição de função no menu suspenso resultante.
Use a caixa de pesquisa para filtrar os resultados para a função desejada. Para este exemplo, pesquise o Colaborador de Dados do Blob de Armazenamento e selecione o resultado correspondente e, em seguida, escolha Avançar.
Em Atribuir acesso a, selecione Usuário, grupo ou entidade de serviço e, em seguida, selecione + Selecionar membros.
No diálogo, pesquise seu nome de usuário do Microsoft Entra (geralmente seu endereço de email user@domain) e escolha Selecionar na parte inferior do diálogo.
Selecione Revisar + atribuir para ir para a página final e, em seguida, Revisar + atribuir novamente para concluir o processo.

Para atribuir uma função no nível do recurso usando a CLI do Azure, primeiro você deve recuperar a ID do recurso usando o comando az storage account show. É possível filtrar as propriedades de saída usando o parâmetro --query.

az storage account show --resource-group '<your-resource-group-name>' --name '<your-storage-account-name>' --query id

Copie a saída Id do comando anterior. Em seguida, você pode atribuir funções usando o comando az role da CLI do Azure.

az role assignment create --assignee "<user@domain>" \
    --role "Storage Blob Data Contributor" \
    --scope "<your-resource-id>"

Para atribuir uma função no nível do recurso usando o Azure PowerShell, primeiro você precisa recuperar a ID do recurso usando o comando Get-AzResource.

Get-AzResource -ResourceGroupName "<yourResourceGroupname>" -Name "<yourStorageAccountName>"

Copie o valor Id da saída do comando anterior. Em seguida, é possível atribuir funções usando o comando New-AzRoleAssignment no PowerShell.

New-AzRoleAssignment -SignInName <user@domain> `
    -RoleDefinitionName "Storage Blob Data Contributor" `
    -Scope <yourStorageAccountId>

Instalar os pacotes para receber eventos

Para o lado receptor, será necessário instalar mais dois pacotes. Neste início rápido, você usará o Armazenamento de Blobs do Azure para persistir pontos de verificação para que o programa não leia os eventos que ele já leu. Ele executa pontos de verificação de metadados em mensagens recebidas em intervalos regulares em um blob. Essa abordagem facilita continuar a receber mensagens mais tarde exatamente do ponto em que você parou.

Sem senha (recomendado)
Cadeia de Conexão

pip install azure-eventhub-checkpointstoreblob-aio
pip install azure-identity

pip install azure-eventhub-checkpointstoreblob-aio

Criar um script do Python para receber eventos

Nesta seção, você criará um script Python para receber eventos do hub de eventos:

Abra seu editor favorito do Python, como o Visual Studio Code.
Crie um script chamado recv.py.

Cole o seguinte código em recv.py:

Sem senha (recomendado)
Cadeia de Conexão

No código, use valores reais para substituir os seguintes espaços reservados:

BLOB_STORAGE_ACCOUNT_URL
BLOB_CONTAINER_NAME
EVENT_HUB_FULLY_QUALIFIED_NAMESPACE
EVENT_HUB_NAME

import asyncio

from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import (
    BlobCheckpointStore,
)
from azure.identity.aio import DefaultAzureCredential

BLOB_STORAGE_ACCOUNT_URL = "BLOB_STORAGE_ACCOUNT_URL"
BLOB_CONTAINER_NAME = "BLOB_CONTAINER_NAME"
EVENT_HUB_FULLY_QUALIFIED_NAMESPACE = "EVENT_HUB_FULLY_QUALIFIED_NAMESPACE"
EVENT_HUB_NAME = "EVENT_HUB_NAME"

credential = DefaultAzureCredential()

async def on_event(partition_context, event):
    # Print the event data.
    print(
        'Received the event: "{}" from the partition with ID: "{}"'.format(
            event.body_as_str(encoding="UTF-8"), partition_context.partition_id
        )
    )

    # Update the checkpoint so that the program doesn't read the events
    # that it has already read when you run it next time.
    await partition_context.update_checkpoint(event)


async def main():
    # Create an Azure blob checkpoint store to store the checkpoints.
    checkpoint_store = BlobCheckpointStore(
        blob_account_url=BLOB_STORAGE_ACCOUNT_URL,
        container_name=BLOB_CONTAINER_NAME,
        credential=credential,
    )

    # Create a consumer client for the event hub.
    client = EventHubConsumerClient(
        fully_qualified_namespace=EVENT_HUB_FULLY_QUALIFIED_NAMESPACE,
        eventhub_name=EVENT_HUB_NAME,
        consumer_group="$Default",
        checkpoint_store=checkpoint_store,
        credential=credential,
    )
    async with client:
        # Call the receive method. Read from the beginning of the partition
        # (starting_position: "-1")
        await client.receive(on_event=on_event, starting_position="-1")

    # Close credential when no longer needed.
    await credential.close()

if __name__ == "__main__":
    # Run the main method.
    asyncio.run(main())

No código, use valores reais para substituir os seguintes espaços reservados:

BLOB_STORAGE_CONNECTION_STRING - Cadeia de conexão com a conta de Armazenamento de Blobs que você anotou anteriormente.
BLOB_CONTAINER_NAME - Nome do contêiner de blob que você criou no armazenamento de blobs.
EVENT_HUB_CONNECTION_STR – Cadeia de conexão com o namespace dos Hubs de Eventos que você anotou anteriormente.
EVENT_HUB_NAME - Nome do hub de eventos.

import asyncio

from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import (
    BlobCheckpointStore,
)

BLOB_STORAGE_CONNECTION_STRING = "BLOB_STORAGE_CONNECTION_STRING"
BLOB_CONTAINER_NAME = "BLOB_CONTAINER_NAME"
EVENT_HUB_CONNECTION_STR = "EVENT_HUB_CONNECTION_STR"
EVENT_HUB_NAME = "EVENT_HUB_NAME"


async def on_event(partition_context, event):
    # Print the event data.
    print(
        'Received the event: "{}" from the partition with ID: "{}"'.format(
            event.body_as_str(encoding="UTF-8"), partition_context.partition_id
        )
    )

    # Update the checkpoint so that the program doesn't read the events
    # that it has already read when you run it next time.
    await partition_context.update_checkpoint(event)

async def main():
    # Create an Azure blob checkpoint store to store the checkpoints.
    checkpoint_store = BlobCheckpointStore.from_connection_string(
        BLOB_STORAGE_CONNECTION_STRING, BLOB_CONTAINER_NAME
    )

    # Create a consumer client for the event hub.
    client = EventHubConsumerClient.from_connection_string(
        EVENT_HUB_CONNECTION_STR,
        consumer_group="$Default",
        eventhub_name=EVENT_HUB_NAME,
        checkpoint_store=checkpoint_store,
    )
    async with client:
        # Call the receive method. Read from the beginning of the
        # partition (starting_position: "-1")
        await client.receive(on_event=on_event, starting_position="-1")

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    # Run the main method.
    loop.run_until_complete(main())

Observação

Para ver exemplos de outras opções para receber eventos do Hub de Eventos de maneira assíncrona usando uma cadeia de conexão, confira a página recv_with_checkpoint_store_async.py do GitHub. Os padrões mostrados também são aplicáveis ao recebimento de eventos sem senha.

Executar o aplicativo receptor

Para executar o script, abra um prompt de comando que tenha Python no caminho e, em seguida, execute este comando:

python recv.py

Executar o aplicativo remetente

Para executar o script, abra um prompt de comando que tenha Python no caminho e, em seguida, execute este comando:

python send.py

A janela do destinatário deve exibir as mensagens que foram enviadas para o hub de eventos.

Solução de problemas

Se você não vir eventos na janela do receptor ou o código relatar um erro, tente as seguintes dicas de solução de problemas:

Se você não vir os resultados de recy.py, execute send.py várias vezes.
Se você vir erros sobre "corrotina" ao usar o código sem senha (com credenciais), verifique se está usando a importação de azure.identity.aio.
Se você vir "Sessão de cliente não revelada" com código sem senha (com credenciais), lembre-se de fechar a credencial quando terminar. Para saber mais, confira Credenciais assíncronas.
Se você vir erros de autorização com recv.py ao acessar o armazenamento, verifique se seguiu as etapas em Criar uma conta de armazenamento do Azure e um contêiner de blobs e atribuiu a função Colaborador de Dados de Blob de Armazenamento à entidade de serviço.
Se você receber eventos com IDs de partição diferentes, esse resultado é esperado. As partições são um mecanismo de organização de dados relacionados ao paralelismo de downstream necessário no consumo de aplicativos. O número de partições em um hub de eventos está diretamente relacionado ao número de leitores simultâneos que você espera ter. Para obter mais informações, confira Saiba mais sobre partições.

Próximas etapas

Neste início rápido, você enviou e recebeu eventos de maneira assíncrona. Para saber como enviar e receber eventos de maneira síncrona, acesse a página sync_samples do GitHub.

Para obter todos os exemplos (síncronos e assíncronos) no GitHub, acesse os exemplos da biblioteca de clientes dos Hubs de Eventos do Azure para Python.

Compartilhar via

Enviar ou receber eventos dos Hubs de Eventos usando o Python

Pré-requisitos

Instalar os pacotes para enviar eventos

Autenticar o aplicativo no Azure

Atribuir funções ao usuário do Microsoft Entra

Funções interna do Azure par Hubs de Eventos do Azure

Obtenha a cadeia de conexão

Enviar eventos

Receber eventos

Criar uma conta de armazenamento e um contêiner de blobs do Azure

Instalar os pacotes para receber eventos

Criar um script do Python para receber eventos

Executar o aplicativo receptor

Executar o aplicativo remetente

Solução de problemas

Próximas etapas

Comentários

Recursos adicionais