Python Nacos demo 实现项目某些参数的自动化配置

工作中用到nacos配置python服务的某些配置。经过一天研究，做了个demo。供大家参考。

demo工程结构

demo工程结构

步骤1，安装nacossdk

#nacos-sdk-python-0.1.6
pip install nacos-sdk-python

步骤2，直接上代码吧。

nacos_helper.py

import nacos
import json_helper

SERVER_ADDRESSES = "10.100.1.1:8848"
NAMESPACE = "bade274b-7bd3-11111111ec810"
USERNAME = "ai_dev_nacos"
PASSWORD = "ogBadfade0M"


ip_list = []
client = nacos.NacosClient(SERVER_ADDRESSES, namespace=NAMESPACE, username=USERNAME, password=PASSWORD)


def callback(args):
    print(type(args["raw_content"]))
    print(args)
    list = json_helper.parse_json_str(args["raw_content"])
    print("args----》", list[0]["cityIds"])
    global ip_list
    ip_list = list[0]["cityIds"]


def addConfigWatcher(data_id, group):
    client.add_config_watcher(data_id=data_id, group=group, cb=callback)
    print("addConfigWatcher success!")


def readLocalConfig(data_id, group):
    try:
        path = "nacos-data/" + "snapshot/" + data_id + "+" + group + "+" + NAMESPACE
        with open(path, mode='r', encoding='utf8') as rf:
            content = rf.read()
    except IOError:
        content = None
        print("Error: 没有找到文件或读取文件失败")

    return content


# 先读取线上配置，失败的话，读取本地配置
def firstGetConfig(data_id, group):
    try:
        config = client.get_config(data_id, group)
    except BaseException:
        config = readLocalConfig(data_id, group)
    if config is not None:
        val = json_helper.parse_json_str(config)
        global ip_list
        ip_list = val[0]["cityIds"]

    return ip_list

nacos_test.py

import time

from nacos_demo import nacos_helper
from nacos_demo import json_helper

NACOS_CONFIG_GROUP = "DEFAULT_GROUP"
AB_TEST_CITY_VERSION_CONFIG = "ab.test.city.version.config"

if __name__ == '__main__':
    val = nacos_helper.firstGetConfig(AB_TEST_CITY_VERSION_CONFIG, NACOS_CONFIG_GROUP)
    print("first config is:", val)
    nacos_helper.addConfigWatcher(AB_TEST_CITY_VERSION_CONFIG, NACOS_CONFIG_GROUP)
    while 1:
        print("other get iplist is:",nacos_helper.ip_list)
        time.sleep(5)
        pass

json_helper.py

import json

def get_json_str(json_value):
    return json.dumps(json_value)

def parse_json_str(json_str, default_data=None):
    try:
        return json.loads(json_str)
    except:
        return default_data

运行效果：

注意：

我在写demo的时候，获取config是没问题的，但是添加watcher，一直报一个错误，“[do-pulling] exception http 403 occur, return empty list, waiting for recovery”，这样的错误。浪费了两个小时，最后发现是我的账号和密码没有添加watcher的权限。改了个有权限的用户名密码就成功监听配置的改变了。

1、http报错 403 一般就是没有访问权限。
2、nacos 获取配置信息是否更新，是通过不断的轮询pull获取的。简单理解就是不停的访问一个接口http://10.100.xxx.xx:8848/nacos/v1/cs/configs/listener?username=xxxxxx&password=xxxxxx

image.png