Inviare o ricevere eventi da hub eventi con Python

Questa guida di avvio rapido illustra come inviare e ricevere eventi da un hub eventi con il pacchetto Python azure-eventhub.

Prerequisiti

Se non si ha familiarità con Hub eventi di Azure, vedere Panoramica di Hub eventi prima di procedere con questa guida di avvio rapido.

Per completare questa guida introduttiva è necessario soddisfare i prerequisiti seguenti:

• Sottoscrizione di Microsoft Azure. Per usare i servizi di Azure, tra cui Hub eventi di Azure, è necessaria una sottoscrizione. Se non si ha un account Azure esistente, iscriversi per ottenere una versione di valutazione gratuita.
• Python 3.8 o versione successiva, con PIP installato e aggiornato.
• Visual Studio Code (scelta consigliata) o qualsiasi altro IDE (Integrated Development Environment).
• Creare uno spazio dei nomi di Hub eventi e un hub eventi. Il primo passaggio consiste nell'usare il portale di Azure per creare uno spazio dei nomi di Hub eventi e ottenere le credenziali di gestione necessarie all'applicazione per comunicare con l'hub eventi. Per creare uno spazio dei nomi e un hub eventi, seguire la procedura descritta in questo articolo.

Installare i pacchetti per l'invio di eventi

Per installare i pacchetti Python per Hub eventi, aprire un prompt dei comandi con Python nel percorso. Modificare la directory nella cartella in cui si desidera mantenere gli esempi.

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

pip install azure-eventhub

Autenticare l'app in Azure

Questo argomento di avvio rapido illustra due modi per connettersi a Hub eventi di Azure: senza password e stringa di connessione. La prima opzione mostra come usare l'entità di sicurezza in Microsoft Entra ID e il controllo degli accessi in base al ruolo per connettersi a uno spazio dei nomi di Hub eventi. Non è necessario preoccuparsi di avere stringa di connessione hardcoded nel codice o in un file di configurazione o in un'archiviazione sicura come Azure Key Vault. La seconda opzione mostra come usare un stringa di connessione per connettersi a uno spazio dei nomi di Hub eventi. Se non si ha familiarità con Azure, è possibile trovare l'opzione stringa di connessione più semplice da seguire. È consigliabile usare l'opzione senza password in applicazioni e ambienti di produzione reali. Per altre informazioni, consultare Autenticazione e autorizzazione. Per altre informazioni sull'autenticazione senza password, vedere la pagina di panoramica.

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

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

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

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

Inviare gli eventi

In questa sezione creare uno script Python per inviare eventi all'hub eventi creato in precedenza.

1. Aprire l'editor preferito di Python, ad esempioVisual Studio Code.

2. Creare uno script denominato send.py. Questo script invia un batch di eventi all'hub eventi creato in precedenza.

3. Incollare il codice seguente in send.py:

   • Senza password (scelta consigliata)
   • Stringa di connessione

   Nel codice usare valori reali per sostituire i segnaposto seguenti:

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

   Nota

   Per esempi di altre opzioni per l'invio di eventi all'hub eventi in modo asincrono usando un stringa di connessione, vedere la pagina send_async.py GitHub. I modelli illustrati sono applicabili anche all'invio di eventi senza password.

Ricevere eventi

In questo argomento di avvio rapido si usa archiviazione BLOB di Azure come archivio di checkpoint. L'archivio di checkpoint viene usato per rendere persistenti i checkpoint (ovvero le ultime posizioni di lettura).

Seguire queste raccomandazioni quando si usa Archiviazione BLOB di Azure come archivio checkpoint:

• Usare un contenitore separato per ogni gruppo di consumer. È possibile usare lo stesso account di archiviazione, ma occorre usare un contenitore per ogni gruppo.
• Non usare il contenitore per altri elementi e non usare l'account di archiviazione per altri elementi.
• L'account di archiviazione deve trovarsi nella stessa area in cui si trova l'applicazione distribuita. Se l'applicazione è locale, provare a scegliere l'area più vicina possibile.

Nella pagina Account di archiviazione del portale di Azure verificare che le impostazioni seguenti siano disabilitate nella sezione Servizio BLOB.

• Spazio dei nomi gerarchico
• Eliminazione temporanea dei BLOB
• Controllo delle versioni

Creare un account di archiviazione di Azure e un contenitore BLOB

Per creare un account di archiviazione di Azure e un contenitore BLOB al suo interno, eseguire queste operazioni:

1. Creare un account di archiviazione di Azure
2. Creare un contenitore BLOB.
3. Eseguire l'autenticazione nel contenitore BLOB.

Assicurarsi di prendere nota della stringa di connessione e del nome del contenitore per un uso successivo nel codice di ricezione.

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

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

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

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

Installare i pacchetti per ricevere gli eventi

Per il lato ricevente, è necessario installare uno o più pacchetti. In questo argomento di avvio rapido si usa archiviazione BLOB di Azure per rendere persistenti i checkpoint, in modo che il programma non legga gli eventi già letti. Vengono eseguiti a intervalli regolari i checkpoint dei metadati sui messaggi ricevuti in un BLOB. Con questo approccio risulta facile continuare a ricevere messaggi dal punto in cui si era interrotto in un momento successivo.

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

pip install azure-eventhub-checkpointstoreblob-aio

Creazione di uno script Python per ricevere gli eventi

In questa sezione viene creato uno script Python per ricevere eventi dall'hub eventi:

1. Aprire l'editor preferito di Python, ad esempioVisual Studio Code.

2. Creare uno script denominato recv.py.

3. Incollare il codice seguente in recv.py:

   • Senza password (scelta consigliata)
   • Stringa di connessione

   Nel codice usare valori reali per sostituire i segnaposto seguenti:

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

   Nota

   Per esempi di altre opzioni per la ricezione di eventi da Hub eventi in modo asincrono usando un stringa di connessione, vedere la pagina recv_with_checkpoint_store_async.py GitHub. I modelli mostrati sono applicabili anche alla ricezione di eventi senza password.

Eseguire l'app ricevente

Per eseguire lo script aprire un prompt dei comandi con un percorso contenente Python, quindi eseguire questo comando:

python recv.py

Eseguire l'app mittente

Per eseguire lo script aprire un prompt dei comandi con un percorso contenente Python, quindi eseguire questo comando:

python send.py

La finestra ricevente dovrebbe visualizzare i messaggi inviati all'hub eventi.

Risoluzione dei problemi

Se non vengono visualizzati eventi nella finestra del ricevitore o il codice segnala un errore, provare i suggerimenti per la risoluzione dei problemi seguenti:

• Se i risultati di recy.py non vengono visualizzati, eseguire send.py più volte.

• Se vengono visualizzati errori relativi alla "coroutine" quando si usa il codice senza password (con credenziali), assicurarsi di usare l'importazione da azure.identity.aio.

• Se viene visualizzato "Sessione client non chiusa" con codice senza password (con credenziali), assicurarsi di chiudere le credenziali al termine. Per altre informazioni, vedere Credenziali asincrone.

• Se vengono visualizzati errori di autorizzazione con recv.py durante l'accesso all'archiviazione, assicurarsi di seguire i passaggi descritti in Creare un account di archiviazione di Azure e un contenitore BLOB e assegnare il ruolo Collaboratore ai dati del BLOB di archiviazione all'entità servizio.

• Se si ricevono eventi con ID di partizione diversi, questo risultato è previsto. Le partizioni sono un meccanismo di organizzazione dei dati correlato al parallelismo downstream necessario per utilizzare le applicazioni. Il numero di partizioni in un hub eventi è direttamente correlato al numero di lettori simultanei previsti. Per altre informazioni, vedere Altre informazioni sulle partizioni.

Passaggi successivi

In questo argomento di avvio rapido sono stati inviati e ricevuti eventi in modo asincrono. Per informazioni su come inviare e ricevere eventi in modo sincrono, passare alla pagina sync_samples di GitHub.

Per tutti gli esempi (sincroni e asincroni) di GitHub, passare alla libreria client di Hub eventi di Azure per gli esempi di Python.