Как запустить асинхронную функцию Azure для триггера Python Event-Hub с помощью класса EventHubConsumerClient?

Я пытаюсь разработать лазурную функцию триггера концентратора событий, которая могла бы получать события из первого концентратора событий и отправлять эти события во второй концентратор событий. В качестве дополнительных функций я хотел бы, чтобы моя функция была асинхронной и использовала контрольные точки хранилища в хранилище BLOB-объектов Azure. Для этого я хотел использовать класс EventHubConsumerClient библиотеки azure-eventhub (https://pypi.org/project/azure-eventhub/ , https://learn.microsoft.com/en- us/javascript/api/@azure/event-hubs/eventhubconsumerclient?view=azure-node-latest)

Однако, похоже, я не могу получать события в первую очередь, когда я тестирую функцию локально в VSCode.

Концентратор событий, который я слушаю, состоит из двух разделов. Его политика общего доступа настроена на отправку и прослушивание. У меня есть небольшой скрипт для отправки ему сообщений для тестирования, и он отлично работает. Моя среда выполнения функции Azure — 4.x с python 3.9.13 с локальной базой conda.

Вот код моей функции для получения событий с классом EventHubConsumerClient в моем init.py:

import logging
import asyncio
import os
from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore
import azure.functions as func

CONNECTION_STR = os.environ.get("EVENT_HUB_CONN_STR")
EVENTHUB_NAME = os.environ.get("EVENT_HUB_NAME")
STORAGE_CONNECTION_STR = os.environ.get("AZURE_STORAGE_CONN_STR")
BLOB_CONTAINER_NAME = os.environ.get("AZURE_STORAGE_NAME")


async def on_event(partition_context, event):
    logging.info("Received event with body: {} from partition: {}.".format(event.body_as_str(encoding = "UTF-8"), partition_context.partition_id))
    await partition_context.update_checkpoint(event)


async def receive(client):
    await client.receive(
        on_event=on_event,
        starting_position = "-1",  # "-1" is from the beginning of the partition.
    )


async def main(one_event: func.EventHubEvent):
    checkpoint_store = BlobCheckpointStore.from_connection_string(STORAGE_CONNECTION_STR, BLOB_CONTAINER_NAME)
    client = EventHubConsumerClient.from_connection_string(
        CONNECTION_STR,
        consumer_group = "$Default",
        eventhub_name=EVENTHUB_NAME,
        checkpoint_store=checkpoint_store,
    )
    async with client:
        await receive(client)


if __name__ == '__main__':
    asyncio.run(main())

источник: https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub/samples/async_samples/recv_with_checkpoint_store_async.py

Примечание. Я знаю, что one_event в main не используется в основном коде, но я хочу, чтобы он вел себя как триггер для запуска main.

Мой файл function.json:

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "type": "eventHubTrigger",
      "name": "one_event",
      "direction": "in",
      "eventHubName": "<My_event_hub_name>",
      "connection": "<My_event_hub_co_str>",
      "cardinality": "one",
      "consumerGroup": "$Default"
    }
  ]
}

Я определил там привязку ввода концентратора событий для использования в качестве триггера.

У меня также есть файл local.settings.json, который содержит некоторые переменные, и файл requirements.txt, в котором, похоже, нет никаких библиотек.

К вашему сведению: я протестировал другой метод (здесь: https://learn.microsoft.com/en-us/azure/azure-functions/functions-bindings-event-hubs-trigger?tabs=in-process%2Cfunctionsv2% 2Cextensionv5&pivots=programming-language-python) (без использования класса EventHubConsumerClient) для получения событий, и он работает нормально, но у меня нет возможностей контрольной точки и асинхронности.

При локальном запуске функции с помощью «func start» вместо получения и печати некоторой базовой информации о полученном событии у меня постоянно печатается много сообщений в моем терминале. Он продолжает печатать и блокирует мой терминал, поэтому я должен вручную убить его и создать новый.

Так что кажется, что мой код не работает должным образом.

* Вероятно, я что-то путаю в методах main() и asyncio.run(). * Вы случайно не знаете в чем может быть проблема? Большое спасибо!