nacos_config.py 2.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071
  1. import json
  2. import platform
  3. from flask import Flask
  4. from nacos import NacosClient
  5. app = Flask(__name__)
  6. config_json = {
  7. "local": {
  8. "address": "192.168.3.80:8848",
  9. "namespace": "dev",
  10. "username": "nacos",
  11. "password": "nacos",
  12. "data_id": "python-mendunr",
  13. "group": "DEFAULT_GROUP"
  14. },
  15. "production": {
  16. "address": "127.0.0.1:8848",
  17. "namespace": "dev",
  18. "username": "nacos",
  19. "password": "2099citu##$$**.com",
  20. "data_id": "python-mendunr",
  21. "group": "DEFAULT_GROUP"
  22. }
  23. }
  24. class NacosConfig:
  25. def __init__(self, environment='local'):
  26. env_config = config_json[environment]
  27. self.SERVER_ADDRESSES = env_config['address']
  28. self.NAMESPACE = env_config['namespace']
  29. self.USERNAME = env_config['username']
  30. self.PASSWORD = env_config['password']
  31. self.data_id = env_config['data_id']
  32. self.group = env_config['group']
  33. self.client = NacosClient(server_addresses=self.SERVER_ADDRESSES, namespace=self.NAMESPACE,
  34. username=self.USERNAME, password=self.PASSWORD)
  35. self.config_data = None
  36. def load_config(self):
  37. """从Nacos加载配置"""
  38. config_str = self.client.get_config(data_id=self.data_id, group=self.group)
  39. self.config_data = json.loads(config_str) if config_str else {}
  40. return self.config_data
  41. def get_config(self):
  42. """返回当前加载的配置数据"""
  43. if self.config_data is None:
  44. self.load_config()
  45. return self.config_data
  46. # 根据操作系统选择环境
  47. plat = platform.system().lower()
  48. environment = 'local' if plat == 'windows' else 'production'
  49. nacos_config = NacosConfig(environment=environment)
  50. configs = nacos_config.get_config()
  51. @app.route('/config')
  52. def get_config_route():
  53. """返回当前配置,或者可以选择在这里重新加载配置"""
  54. # 可以在这里选择是否重新加载配置,或者根据需要进行其他逻辑处理
  55. # nacos_config.load_config() # 如果需要每次请求都检查更新,可以取消注释这行代码
  56. return json.dumps(nacos_config.get_config())
  57. if __name__ == '__main__':
  58. app.run(debug=True)