如何将 Azure 事件网格中的事件发布到 python 中的主题？

Question

无法在我的 VS Code 中导入 eventgrid 模块，我已经在 requirement.txt 中添加了所有模块并从我的 cmd 安装了 pip。

此外，下面是我正在寻找并尝试使用的 python 函数：

def publish_event(): 
    # authenticate client 
    credential = AzureKeyCredential(key) 
    client = EventGridPublisherClient(endpoint, credential) 
 
    custom_schema_event = { 
        "customSubject": "sample", 
        "customEventType": "sample.event", 
        "customDataVersion": "2.0", 
        "customId": uuid.uuid4(), 
        "customEventTime": dt.datetime.now(UTC()).isoformat(), 
        "customData": "sample data" 
    } 
 
    # publish events 
    for _  in range(3): 
 
        event_list = []     # list of events to publish 
        # create events and append to list 
        for j in range(randint(1, 3)): 
            event_list.append(custom_schema_event) 
 
        # publish list of events 
        client.send(event_list) 
        print("Batch of size {} published".format(len(event_list))) 
        time.sleep(randint(1, 5)) 
 
 
if name == '__main__': 
    publish_event()

我不确定这是否是实现此目的的正确方法，正在寻找解决模块问题和发布事件的更好方法。

帮助将不胜感激！

Answer 1

从你的问题中我们可以看出有两点需要回答：

Python 软件包安装问题
事件网格主题问题

使用pip install安装Azure Python包，我们需要在终端开启虚拟环境，命令如下：
```
   python -m venv .venv
  .venv\Scripts\activate
```
启用 venv 后，在 requirements.txt 中添加所有 eventgrid、storage 和其他与 azure 相关的包，然后运行在 cmd 中执行以下命令：

pip install -r requirements.txt

现在，查看您的代码，函数看起来不错，因为它将事件发送到一般主题。以类似的方式，我们有一个示例，它让我们对将事件网格事件发布到自定义主题有一些了解。这在 Azure-SDK-Python-GitHub 存储库中进行了解释。 Python 代码如下：

  import os
  from random import randint, sample
  import time

  from azure.core.credentials import AzureKeyCredential
  from azure.eventgrid import EventGridPublisherClient, EventGridEvent

  key = os.environ["EG_ACCESS_KEY"]
  endpoint = os.environ["EG_TOPIC_HOSTNAME"]

  #authenticate client
  credential = AzureKeyCredential(key)
  client = EventGridPublisherClient(endpoint, credential)
  services = ["EventGrid", "ServiceBus", "EventHubs", "Storage"]    #possible values for data field

  def publish_event():
      #publish events
      for _ in range(3):

          event_list = []     #list of events to publish
          #create events and append to list
          for j in range(randint(1, 3)):
              sample_members = sample(services, k=randint(1, 4))      #select random subset of team members
              event = EventGridEvent(
                      subject="Door1",
                      data={"team": sample_members},
                      event_type="Azure.Sdk.Demo",
                      data_version="2.0"
                      )
              event_list.append(event)

          #publish list of events
          client.send(event_list)
          print("Batch of size {} published".format(len(event_list)))
          time.sleep(randint(1, 5))

  if __name__ == '__main__':
      publish_event()

如何将 Azure 事件网格中的事件发布到 python 中的主题？

How to publish events in azure event grid to a topic in python?

python

azure-eventgrid

azure-python-sdk