如何用Python向Azure Event Grid主题发布事件及解决模块导入问题

• 确认VS Code Python解释器匹配：VS Code右下角展示的Python版本，需和你执行pip安装命令的环境完全一致。可在VS Code内置终端执行pip --version查看pip对应的环境路径，执行pip show azure-eventgrid确认包是否安装在当前环境下。如果环境不匹配，切换解释器后重新执行安装命令即可。
• 核对依赖包配置：requirements.txt中需填写正确的包名azure-eventgrid，建议指定稳定版本避免兼容问题，示例配置：azure-eventgrid>=4.15.0，可执行pip install --upgrade azure-eventgrid强制更新到最新稳定版。
• 补全正确的导入语句：代码开头必须添加全量导入声明，示例如下：

import uuid
import datetime as dt
import time
from random import randint
from azure.core.credentials import AzureKeyCredential
from azure.eventgrid import EventGridPublisherClient

原来的代码存在几个问题：入口判断name缺少双下划线、循环中复用同一个事件对象会导致所有事件id和时间完全相同、缺少异常捕获逻辑、未声明必填的endpoint和key参数。优化后的完整可运行代码如下：

import uuid
import datetime as dt
import time
from random import randint
from azure.core.credentials import AzureKeyCredential
from azure.eventgrid import EventGridPublisherClient

# 配置参数，可根据实际情况替换
EVENTGRID_ENDPOINT = "你的事件网格端点地址"
EVENTGRID_ACCESS_KEY = "你的访问密钥"

def publish_event():
    # 客户端鉴权
    credential = AzureKeyCredential(EVENTGRID_ACCESS_KEY)
    client = EventGridPublisherClient(EVENTGRID_ENDPOINT, credential)

    # 发布3批事件
    for _ in range(3):
        event_list = []
        # 每批随机生成1-3个事件
        for j in range(randint(1, 3)):
            # 每次生成新的事件，保证id、时间唯一
            custom_schema_event = {
                "customSubject": "sample",
                "customEventType": "sample.event",
                "customDataVersion": "2.0",
                "customId": str(uuid.uuid4()),
                "customEventTime": dt.datetime.now(dt.timezone.utc).isoformat(),
                "customData": "sample data"
            }
            event_list.append(custom_schema_event)

        try:
            # 批量发布事件
            client.send(event_list)
            print(f"Batch of size {len(event_list)} published")
        except Exception as e:
            print(f"Failed to publish batch: {str(e)}")
        # 随机间隔1-5秒
        time.sleep(randint(1, 5))

if __name__ == '__main__':
    publish_event()

主要优化点：

• 修复了入口判断的拼写错误，补充了必填配置参数的声明
• 每次循环生成全新的事件对象，避免所有事件共用相同的id和触发时间
• 新增异常捕获逻辑，可处理网络波动、鉴权失败、参数错误等常见异常
• 替换了废弃的UTC()写法，使用标准的dt.timezone.utc获取UTC时间
• uuid转成字符串格式，避免发送时出现类型不兼容问题

内容的提问来源于stack exchange，提问作者Rishika