Partager via

Démarrage rapide : Envoyer des événements à des hubs d’événements ou recevoir des événements à l’aide de Python

Dans ce guide de démarrage rapide, vous allez apprendre à envoyer et recevoir des événements à partir d’un hub d’événements à l’aide du package Python azure-eventhub .

Prérequis

Si vous débutez avec Azure Event Hubs, consultez la vue d’ensemble d’Event Hubs avant de suivre ce guide de démarrage rapide.

Pour suivre ce guide de démarrage rapide, vérifiez que vous disposez des conditions préalables suivantes :

  • Abonnement Microsoft Azure : inscrivez-vous à une version d’évaluation gratuite si vous n’en avez pas.
  • Python 3.8 ou version ultérieure : vérifiez que pip est installé et mis à jour.
  • Visual Studio Code (recommandé) : ou utilisez n’importe quel autre IDE de votre choix.
  • Espace de noms Event Hubs et hub d’événements : suivez ce guide pour les créer dans le portail Azure.

Installer les packages pour envoyer des événements

Pour installer les packages Python pour Event Hubs, ouvrez une invite de commandes qui a Python dans son chemin. Remplacez le répertoire par le dossier dans lequel vous souhaitez conserver vos exemples.

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

Authentifier l’application sur Azure

Ce guide de démarrage rapide vous montre deux façons de vous connecter à Azure Event Hubs :

  • Sans mot de passe. Utilisez votre principal de sécurité dans Microsoft Entra ID et le contrôle d’accès en fonction du rôle (RBAC) pour vous connecter à un espace de noms Event Hubs. Vous n’avez pas besoin de vous soucier d’avoir des chaînes de connexion codées en dur dans votre code, dans un fichier de configuration ou dans un stockage sécurisé comme Azure Key Vault.
  • Chaîne de connexion. Utilisez une chaîne de connexion pour vous connecter à un espace de noms Event Hubs. Si vous débutez dans Azure, vous trouverez peut-être l’option de chaîne de connexion plus facile à suivre.

Nous vous recommandons d’utiliser l’option sans mot de passe dans les applications réelles et les environnements de production. Pour plus d’informations, consultez l’authentification et l’autorisation Service Bus et les connexions sans mot de passe pour les services Azure.

Attribuer des rôles à votre utilisateur Microsoft Entra

Lorsque vous développez localement, assurez-vous que le compte d’utilisateur qui se connecte à Azure Event Hubs dispose des autorisations appropriées. Vous avez besoin du rôle Propriétaire des données Azure Event Hubs pour envoyer et recevoir des messages. Pour vous attribuer ce rôle, vous avez besoin du rôle Administrateur de l’accès utilisateur ou d’un autre rôle qui inclut l’action Microsoft.Authorization/roleAssignments/write . Vous pouvez attribuer des rôles RBAC Azure à un utilisateur en utilisant le Portail Azure, Azure CLI ou Azure PowerShell. Pour plus d’informations, consultez la page Comprendre l’étendue de RBAC Azure.

L’exemple suivant attribue le rôle Azure Event Hubs Data Owner à votre compte d’utilisateur, qui fournit un accès complet aux ressources Azure Event Hubs. Dans un scénario réel, suivez le principe des privilèges minimum pour accorder aux utilisateurs uniquement les autorisations minimales nécessaires à un environnement de production plus sécurisé.

Rôles intégrés Azure pour Azure Event Hubs

Pour Azure Event Hubs, la gestion des espaces de noms et de toutes les ressources associées via le portail Azure et l’API de gestion des ressources Azure est déjà protégée à l’aide du modèle RBAC Azure. Azure fournit les rôles intégrés suivants pour autoriser l’accès à un espace de noms Event Hubs :

Si vous souhaitez créer un rôle personnalisé, consultez Droits requis pour les opérations Service Bus.

Important

Dans la plupart des cas, la propagation de l’attribution de rôle dans Azure peut prendre une ou deux minutes. Dans de rares cas, cela peut prendre jusqu’à huit minutes. Si vous recevez des erreurs d’authentification lorsque vous exécutez votre code pour la première fois, patientez quelques instants et réessayez.

  1. Dans le portail Azure, recherchez votre espace de noms Event Hubs à l’aide de la barre de recherche principale ou de la navigation gauche.

  2. Dans la page vue d’ensemble, sélectionnez Contrôle d’accès (IAM) dans le menu de gauche.

  3. Dans la page Contrôle d’accès (IAM), sélectionnez l’onglet Attributions de rôle.

  4. Sélectionnez + Ajouter dans le menu supérieur. Sélectionnez Ensuite Ajouter une attribution de rôle.

    Capture d’écran montrant comment attribuer un rôle.

  5. Utilisez la zone de recherche pour filtrer les résultats selon le rôle souhaité. Pour cet exemple, recherchez Azure Event Hubs Data Owner et sélectionnez le résultat correspondant. Ensuite, choisissez Suivant.

  6. Sous Attribuer l’accès à, sélectionnez Utilisateur, groupe ou principal de service. Choisissez ensuite + Sélectionner des membres.

  7. Dans la boîte de dialogue, recherchez votre nom d’utilisateur Microsoft Entra (généralement votre adresse e-mail user@domain ). Choisissez Sélectionner en bas de la boîte de dialogue.

  8. Sélectionnez Vérifier + affecter pour accéder à la page finale. Sélectionnez Vérifier + attribuer à nouveau pour terminer le processus.

Envoyer des événements

Dans cette section, vous allez créer un script Python pour envoyer des événements au hub d’événements que vous avez créé.

  1. Ouvrez votre éditeur Python favori, tel que Visual Studio Code.

  2. Créez un script appelé send.py. Ce script envoie un lot d’événements au hub d’événements que vous avez créé.

  3. Collez le code suivant dans send.py :

    Dans le code, utilisez des valeurs réelles pour remplacer les espaces réservés suivants :

    • EVENT_HUB_FULLY_QUALIFIED_NAMESPACE - Vous voyez le nom complet qualifié sur la page Vue d’ensemble de l’espace de noms. Il doit être au format : <NAMESPACENAME>>.servicebus.windows.net.
    • EVENT_HUB_NAME - Nom de l’Event Hub.
    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,
    )
    print("Producer client created successfully.") 
    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())

    Remarque

    Pour obtenir des exemples d’autres options d’envoi d’événements à un hub d’événements de manière asynchrone à l’aide d’une chaîne de connexion, consultez la page send_async.py GitHub. Les modèles indiqués s’appliquent également à l’envoi d’événements sans mot de passe.

Recevoir des événements

Ce guide de démarrage rapide utilise un stockage Blob Azure comme magasin de points de contrôle. Le magasin de points de contrôle est utilisé pour conserver les points de contrôle (autrement dit, les dernières positions de lecture).

Suivez ces recommandations lorsque vous utilisez le Stockage Blob Azure en tant que magasin de points de contrôle :

  • Utilisez un conteneur distinct pour chaque groupe de consommateurs. Vous pouvez utiliser le même compte de stockage, mais utilisez un conteneur par groupe.
  • N’utilisez pas le compte de stockage pour autre chose.
  • N’utilisez pas le conteneur pour tout autre chose.
  • Créez le compte de stockage dans la même région que l’application déployée. Si l’application est locale, essayez de choisir la région la plus proche possible.

Sur la page Compte de stockage du Portail Azure, dans la section Service BLOB, vérifiez que les paramètres suivants sont désactivés.

  • Espace de noms hiérarchique
  • Suppression réversible de blob
  • Contrôle de version

Créer un compte de stockage Azure et un conteneur d’objets blob

Créez un compte de stockage Azure et un conteneur d’objets blob dans celui-ci en effectuant les étapes suivantes :

  1. Création d’un compte de stockage Azure
  2. Créez un conteneur d’objets blob.
  3. S’authentifier auprès du conteneur d’objets blob.

Veillez à enregistrer la chaîne de connexion et le nom du conteneur en vue de les utiliser dans le code de réception.

Lorsque vous développez localement, assurez-vous que le compte d’utilisateur qui accède aux données d’objet blob dispose des autorisations appropriées. Pour lire et écrire des données d’objet blob, vous avez besoin de Contributeur aux données Blob du stockage. Pour vous attribuer ce rôle, vous devez disposer du rôle Administrateur de l’accès utilisateur ou d’un autre rôle qui inclut l’action Microsoft.Authorization/roleAssignments/write . Vous pouvez attribuer des rôles RBAC Azure à un utilisateur en utilisant le Portail Azure, Azure CLI ou Azure PowerShell. Pour plus d’informations, consultez Comprendre l’étendue du contrôle d’accès en fonction du rôle (RBAC).

Dans ce scénario, vous attribuez des autorisations à votre compte d’utilisateur, délimitées au compte de stockage, pour suivre le principe du privilège minimum. Cette pratique offre aux utilisateurs uniquement les autorisations minimales nécessaires et crée des environnements de production plus sécurisés.

L’exemple suivant attribue le rôle Contributeur aux données blob de stockage à votre compte d’utilisateur, qui fournit à la fois un accès en lecture et en écriture aux données blob dans votre compte de stockage.

Important

Dans la plupart des cas, la propagation de l’attribution de rôle dans Azure peut prendre une ou deux minutes. Dans de rares cas, cela peut prendre jusqu’à huit minutes. Si vous recevez des erreurs d’authentification lorsque vous exécutez votre code pour la première fois, patientez quelques instants et réessayez.

  1. Dans le Portail Azure, recherchez votre compte de stockage à l’aide de la barre de recherche principale ou de la navigation gauche.

  2. Dans la page du compte de stockage, sélectionnez Contrôle d’accès (IAM) dans le menu de gauche.

  3. Dans la page Contrôle d’accès (IAM), sélectionnez l’onglet Attributions de rôle.

  4. Sélectionnez + Ajouter dans le menu supérieur. Sélectionnez Ensuite Ajouter une attribution de rôle.

    Capture d’écran montrant comment attribuer un rôle de compte de stockage.

  5. Utilisez la zone de recherche pour filtrer les résultats selon le rôle souhaité. Pour cet exemple, recherchez Contributeur aux données Blob du stockage. Sélectionnez le résultat correspondant, puis choisissez Suivant.

  6. Sous Attribuer l’accès à, sélectionnez Utilisateur, groupe ou principal de service, puis sélectionnez + Sélectionner des membres.

  7. Dans la boîte de dialogue, recherchez votre nom d’utilisateur Microsoft Entra (généralement votre adresse e-mail user@domain), puis choisissez Sélectionner en bas de la boîte de dialogue.

  8. Sélectionnez Vérifier + affecter pour accéder à la page finale. Sélectionnez Vérifier + attribuer à nouveau pour terminer le processus.

Installer les packages pour recevoir des événements

Pour le côté récepteur, vous avez besoin d’installer un autre package, ou plus. Dans ce guide de démarrage rapide, vous utilisez le stockage Blob Azure pour conserver les points de contrôle afin que le programme ne lit pas les événements qu’il lit déjà. Il effectue des points de contrôle de métadonnées sur les messages reçus à intervalles réguliers dans un blob. Cette approche permet ultérieurement de continuer à recevoir des messages à partir du point où vous vous étiez arrêté.

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

Créer un script Python pour recevoir des événements

Dans cette section, vous allez créer un script Python pour recevoir des événements de votre Event Hub :

  1. Ouvrez votre éditeur Python favori, tel que Visual Studio Code.

  2. Créez un script appelé recv.py.

  3. Collez le code suivant dans recv.py :

    Dans le code, utilisez des valeurs réelles pour remplacer les espaces réservés suivants :

    • BLOB_STORAGE_ACCOUNT_URL - Cette valeur doit être au format : https://<YOURSTORAGEACCOUNTNAME>.blob.core.windows.net/
    • BLOB_CONTAINER_NAME - Nom du conteneur d’objets blob dans le compte de stockage Azure.
    • EVENT_HUB_FULLY_QUALIFIED_NAMESPACE - Vous voyez le nom complet qualifié sur la page Vue d’ensemble de l’espace de noms. Il doit être au format : <NAMESPACENAME>>.servicebus.windows.net.
    • EVENT_HUB_NAME - Nom de l’Event Hub.
    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())

    Remarque

    Pour obtenir des exemples d’autres options de réception d’événements à partir d’un hub d’événements de manière asynchrone à l’aide d’une chaîne de connexion, consultez la page recv_with_checkpoint_store_async.py GitHub. Les modèles indiqués s’appliquent également à la réception d’événements sans mot de passe.

Exécuter l’application réceptrice

  1. Lancer une invite de commandes.

  2. Exécutez la commande suivante et connectez-vous à l’aide du compte ajouté au rôle Propriétaire des données Azure Event Hubs sur l’espace de noms Event Hubs et le rôle Contributeur aux données blob de stockage sur le compte de stockage Azure.

    az login

  3. Basculez vers le dossier contenant le fichier receive.py, puis exécutez la commande suivante :

    python recv.py

Exécuter l’application de l’expéditeur

  1. Lancer une invite de commandes.

  2. Exécutez la commande suivante et connectez-vous à l’aide du compte ajouté au rôle Propriétaire des données Azure Event Hubs sur l’espace de noms Event Hubs et le rôle Contributeur aux données blob de stockage sur le compte de stockage Azure.

    az login

  3. Basculez vers le dossier qui a le send.py, puis exécutez cette commande :

    python send.py

La fenêtre réceptrice doit afficher les messages qui ont été envoyés au hub d’événements.

Dépannage

Si vous ne voyez pas d’événements dans la fenêtre du récepteur ou si le code signale une erreur, essayez les conseils de dépannage suivants :

  • Si vous ne voyez pas les résultats de recy.py, exécutez send.py plusieurs fois.

  • Si vous voyez des erreurs concernant « coroutine » lors de l’utilisation du code sans mot de passe (avec des informations d’identification), vérifiez que vous utilisez l’importation à partir de azure.identity.aio.

  • Si vous voyez « Session client non fermée » avec du code sans mot de passe (avec des informations d’identification), veillez à fermer les informations d’identification lorsque vous avez terminé. Pour plus d’informations, consultez Informations asynchrones.

  • Si vous voyez des erreurs d’autorisation avec recv.py lors de l’accès au stockage, vérifiez que vous avez suivi les étapes décrites dans Créer un compte de stockage Azure et un conteneur d’objets blob et que vous avez attribué le rôle Contributeur aux données blob de stockage au principal de service.

  • Si vous recevez des événements avec des ID de partition différents, ce résultat est attendu. Les partitions constituent un mécanisme d’organisation des données. Elles sont liées au degré de parallélisme en aval requis lors de la consommation des applications. Le choix du nombre de partitions dans un concentrateur d’événements est directement lié au nombre de lecteurs simultanés que vous prévoyez d’avoir. Pour plus d’informations, consultez En savoir plus sur les partitions.

Étapes suivantes

Dans ce guide de démarrage rapide, vous avez envoyé et reçu des événements de manière asynchrone. Pour savoir comment envoyer et recevoir des événements de manière synchrone, accédez à la page GitHub sync_samples.

Explorez d’autres exemples et scénarios avancés dans la bibliothèque cliente Azure Event Hubs pour les exemples Python.

  • Last updated on