如何使用Python查询企业版Consul指定namespace的服务及元数据

前置准备 #

首先安装适配Consul API的Python依赖包：

pip install python-consul

注：python-consul库原生支持传递企业版Consul的namespace参数，无需额外二次封装API。

核心实现 #

逻辑说明 #

• 输入参数仅支持ns_1、ns_2两个合法namespace，非法输入会直接抛出参数校验异常
• 通过Consul Catalog接口查询指定namespace下的服务列表，再遍历获取单个服务的实例详情及关联元数据
• 所有查询请求自动携带namespace标识，避免跨namespace数据混淆

示例代码 #

import consul

def get_consul_services_by_namespace(target_ns: str) -> dict:
    # 输入参数合法性校验
    valid_ns = ["ns_1", "ns_2"]
    if target_ns not in valid_ns:
        raise ValueError(f"非法namespace参数，仅支持可选值：{','.join(valid_ns)}")
    
    # 初始化Consul客户端，替换为你实际的Consul配置
    consul_client = consul.Consul(
        host="你的Consul服务地址",
        port=8500,
        token="你的Consul访问Token（未开启ACL可省略）",
        scheme="http" # 开启HTTPS则修改为https
    )
    
    # 查询指定namespace下的所有服务列表
    _, services_list = consul_client.catalog.services(ns=target_ns)
    result = {}
    
    # 遍历查询每个服务的实例详情和元数据
    for service_name in services_list.keys():
        _, service_instances = consul_client.catalog.service(service_name, ns=target_ns)
        result[service_name] = service_instances
    
    return result

# 调用示例
if __name__ == "__main__":
    # 查询ns_1的服务数据
    ns1_data = get_consul_services_by_namespace("ns_1")
    print("ns_1下的服务及元数据：", ns1_data)
    # 查询ns_2的服务数据
    ns2_data = get_consul_services_by_namespace("ns_2")
    print("ns_2下的服务及元数据：", ns2_data)

结果说明 #

返回结果为嵌套字典结构：

• 一级key为对应namespace下的服务名称
• 每个服务对应的值为实例列表，单实例包含的核心字段：
  • Address：实例部署地址
  • ServicePort：服务暴露端口
  • ServiceMeta：服务绑定的自定义元数据
  • ServiceTags：服务关联的标签列表

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