python - 如何将天蓝色事件网格中的事件发布到python中的主题？

Question

无法在我的 VS 代码中导入 eventgrid 模块，我已在 require.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

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

1. Python包安装问题
2. Eventgrid主题问题

• 要使用 pip install 安装 Azure Python 包，我们需要在终端中启用虚拟环境，命令如下：

     python -m venv .venv
    .venv\Scripts\activate
  

  启用 venv 后，在 requirements.txt 中添加所有 eventgrid、storage 和其他 azure 相关的包，然后在 cmd 中运行以下命令：

  pip install -r requirements.txt

• 现在，查看您的代码，函数看起来很好，因为它将事件发送到一般主题。以类似的方式，我们有一个示例，它为我们提供了有关将 eventgrid 事件发布到自定义主题的一些见解。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()