配置导出脚本v1 api

[chanjetsdp]

参考接口文档：https://nacos.io/zh-cn/docs/open-api.html

import os

import requests


class MyNacosClient():
    def __init__(self, server_url, username, password):
        self.server_url = server_url
        url = server_url + '/nacos/v1/auth/login'
        payload = {'username': username, 'password': password}
        response = requests.post(url, data=payload)
        if response.status_code == 200:
            auth_info = response.json()
            access_token = auth_info['accessToken']
            print(f"Access Token: {access_token}")
            self.login_param = {
                "accessToken": access_token
            }
        else:
            print(f"Error: {response.text}")
            raise Exception('不能获取到token')

    def get_namespace(self):
        url = self.server_url + '/nacos/v1/console/namespaces'
        params = {}
        params.update(self.login_param)

        response = requests.get(url, params=params)
        if response.status_code == 200 and response.json()['code'] == 200:
            return response.json()['data']

    def get_config_list(self, tenant):
        url = self.server_url + '/nacos/v1/cs/configs'
        params = {"dataId": '',
                  'group': '',
                  'appName': '',
                  'tenant': tenant,
                  'pageNo': '1',
                  'pageSize': '10000',
                  'search': 'accurate',
                  }
        params.update(self.login_param)
        print('{}?{}'.format(url, '&'.join(['{}={}'.format(k, v) for k, v in params.items()])))
        response = requests.get(url, params=params)
        if response.status_code == 200:
            return response.json()['pageItems']

    def get_config(self, tenant, group, dataId):
        url = self.server_url + '/nacos/v1/cs/configs'
        params = {"tenant": tenant,
                  "group": group,
                  "dataId": dataId
                  }

        params.update(self.login_param)
        print('{}?{}'.format(url, '&'.join(['{}={}'.format(k, v) for k, v in params.items()])))

        response = requests.get(url, params=params)
        if response.status_code == 200:
            return response.text


if __name__ == '__main__':
    my_client = MyNacosClient(server_url="http://your.nacos.server.url",
                              username='YOUR_USERNAME',
                              password='YOUR_PASSWORD'
                              )
    for namespace_item in my_client.get_namespace():
        for config_item in my_client.get_config_list(namespace_item['namespace']):
            config_text = my_client.get_config(config_item['tenant'], config_item['group'], config_item['dataId'])

            output_file = 'resources/ns_{}__g_{}__d_{}'.format(
                namespace_item['namespaceShowName'],
                # config_item['tenant'],
                config_item['group'],
                config_item['dataId']

            )
            os.makedirs(os.path.dirname(output_file), exist_ok=True)
            with open(output_file, 'w') as fw:
                fw.write(config_text)