Nacos服务部署及配置读取

docker pull nacos/nacos-server

docker run -d --name nacos -p 8848:8848 -e PREFER_HOST_MODE=hostname -e MODE=standalone nacos/nacos-server

# -*- coding: utf-8 -*-
import configparser
import json
import logging
import requests
import yaml


class NacosCodeException(Exception):
    """Nacos 接口返回异常状态码"""

    def __init__(self, value: str):
        self.value = value

    def __str__(self):
        return f"Nacos 接口返回状态码不等于200, status_code={self.value}"


class NacosLoadException(Exception):
    """ Nacos 数据结构异常"""

    def __init__(self, data_id: str, group: str, conf_type: str):
        self.data_id = data_id
        self.group = group
        self.conf_type = conf_type

    def __str__(self):
        return f"Nacos 数据结构错误, data_id={self.data_id}, " 
               f"group={self.group}, conf_type={self.conf_type}"


def nacos_conf_load(data_id: str, group: str, conf_type: str):
    """从Nacos读取配置"""
    assert conf_type in ("json", "ini", "yaml")

    params = {
        "dataId": data_id,
        "group": group
    }
    r = requests.get("http://127.0.0.1:8848/nacos/v1/cs/configs", params=params)
    if r.status_code != 200:
        raise NacosCodeException(str(r.status_code))
    # 加载配置信息
    try:
        if conf_type == "json":
            config = json.loads(r.text)
        elif conf_type == "ini":
            cf = configparser.RawConfigParser()
            cf.read_string(r.text)
            config = {}
            config.update(cf)
        else:
            config = yaml.load(r.text, Loader=yaml.SafeLoader)
    except Exception:
        raise NacosLoadException(data_id, group, conf_type)
    return config

if __name__ == "__main__":
	# nacos连通测试用
    conf = nacos_conf_load("test11", "DEFAULT_GROUP", "yaml")
    print(conf)