如何使用Python获取IoT Hub中的路由消息详情?
获取Azure IoT Hub路由详情的Python方案
要获取IoT Hub消息路由选项卡表格里的全部内容,直接用官方的azure-mgmt-iothub SDK即可,这是专门用于管理Azure IoT Hub资源的Python包。
步骤1:安装SDK
执行以下命令安装依赖:
pip install azure-mgmt-iothub azure-identity
azure-identity用于完成Azure身份认证,支持本地开发、Azure环境等多种场景。
步骤2:编写代码获取路由信息
以下代码可获取路由规则、关联端点等表格内的所有核心内容:
from azure.identity import DefaultAzureCredential from azure.mgmt.iothub import IotHubClient # 替换为你的Azure资源信息 SUBSCRIPTION_ID = "你的订阅ID" RESOURCE_GROUP = "你的资源组名称" IOT_HUB_NAME = "你的IoT Hub实例名称" # 初始化客户端 credential = DefaultAzureCredential() client = IotHubClient(credential, SUBSCRIPTION_ID) # 获取所有路由规则(对应路由选项卡的规则列表) routing_rules = client.iothub_resource.get_routing_rules(RESOURCE_GROUP, IOT_HUB_NAME) print("=== 路由规则详情 ===") for rule in routing_rules.value: print(f"路由名称: {rule.name}") print(f"数据源: {rule.data_source}") print(f"关联端点: {', '.join(rule.endpoint_names)}") print(f"路由过滤条件: {rule.condition if rule.condition else '无'}") print(f"状态: {rule.status}") print("-------------------") # 获取所有路由端点信息(对应规则关联的目标端点详情) routing_endpoints = client.iothub_resource.get_routing_endpoints(RESOURCE_GROUP, IOT_HUB_NAME) print("\n=== 路由端点详情 ===") # 遍历不同类型的端点(事件中心、存储、服务总线等) if routing_endpoints.event_hub_endpoints: for name, endpoint in routing_endpoints.event_hub_endpoints.items(): print(f"事件中心端点: {name}") print(f"目标资源ID: {endpoint.id}") print("-------------------") if routing_endpoints.storage_container_endpoints: for name, endpoint in routing_endpoints.storage_container_endpoints.items(): print(f"存储容器端点: {name}") print(f"目标存储账户: {endpoint.storage_account_resource_group}/{endpoint.storage_account_name}") print(f"容器名称: {endpoint.container_name}") print("-------------------")
说明
- 该SDK可覆盖路由选项卡表格里的所有字段:路由名称、数据源、目标端点、过滤条件、状态,以及端点的详细配置信息。
- 身份认证方面,
DefaultAzureCredential会自动读取本地环境的Azure认证信息(比如Azure CLI登录、VS Code Azure账户等),生产环境可改用服务主体认证。
内容的提问来源于stack exchange,提问作者jakeMantle
相关产品推荐
相关产品推荐

