使用Python管理Azure Event Hubs的类导入与操作实现问题

核心修正点 #

• 无需手动导入、实例化EventHubsOperations类：EventHubManagementClient实例内置的event_hubs属性就是已经完成初始化的EventHubsOperations对象，直接调用该属性下的方法即可实现事件中心的创建、删除操作。
• 原代码中EventHubManagementClient实例化参数错误：构造方法第一个入参应为身份凭证，第二个为订阅ID，不需要传入Key Vault地址。

可运行完整代码 #

import setenv
import os
from azure.mgmt.eventhub import EventHubManagementClient
from azure.identity import DefaultAzureCredential

# 加载环境变量
setenv.import_env_vars('')

# 基础参数配置，按实际情况替换
subscription_id = os.environ["AZURE_SUBSCRIPTION_ID"]
resource_group = "你的资源组名称"
eh_namespace = "你的Event Hub命名空间名称"
target_eventhub_name = "目标事件中心名称"

credential = DefaultAzureCredential()

# 实例化管理客户端
print('Creating EH_client...')
EH_client = EventHubManagementClient(credential, subscription_id)
print('Created.')

# 事件中心创建示例
create_params = {
    "partition_count": 4,
    "message_retention_in_days": 1
}
create_result = EH_client.event_hubs.create_or_update(
    resource_group_name=resource_group,
    namespace_name=eh_namespace,
    event_hub_name=target_eventhub_name,
    parameters=create_params
)
print(f"事件中心创建成功，详情：{create_result.as_dict()}")

# 事件中心删除示例，需要时解除注释即可
# EH_client.event_hubs.delete(
#     resource_group_name=resource_group,
#     namespace_name=eh_namespace,
#     event_hub_name=target_eventhub_name
# )
# print("事件中心删除成功")

原问题解答 #

1. EventHubsOperations导入路径 #

若确实需要手动导入（常规开发场景不需要），导入语句为：
from azure.mgmt.eventhub.v2021_01_01_preview.operations import EventHubsOperations

2. 实例化参数说明 #

手动实例化需要的三个参数都来自已初始化的EventHubManagementClient实例，对应为EH_client._config、EH_client._serialize、EH_client._deserialize。该操作为SDK内部实现逻辑，普通开发者无需手动处理，直接使用内置的event_hubs属性即可。

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