diff --git a/application/common/__init__.py b/application/common/__init__.py index 004634da5dcf20580c320129936483ea4ab00642..ab260d70dc3101e5fe2eb712a1d0d7d290977966 100644 --- a/application/common/__init__.py +++ b/application/common/__init__.py @@ -11,4 +11,5 @@ """ from .config import ConfigHelper +from .env import EnvVarHelper from .file import FileHelper diff --git a/application/common/env.py b/application/common/env.py new file mode 100644 index 0000000000000000000000000000000000000000..45d8223868af5c1073e41bed6f1e909452035b12 --- /dev/null +++ b/application/common/env.py @@ -0,0 +1,44 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +# @Version : Python 3.11.4 +# @Software : Sublime Text 4 +# @Author : StudentCWZ +# @Email : StudentCWZ@outlook.com +# @Date : 2023/11/4 11:15 +# @File : env.py +# @Description : +""" + +import os +from typing import Union + + +class EnvVarHelper: + """ + A helper class for handling environment variables. + """ + + @property + def consul_vars(self) -> dict: + """ + Get the consul-related environment variables. + + :return: a dict containing the consul-related environment variables. + """ + return { + 'host': os.environ.get('CONSUL_HOST', 'localhost'), + 'port': os.environ.get('CONSUL_PORT', 8500), + 'dc': os.environ.get('CONSUL_DC', 'dc1'), + 'token': os.getenv('CONSUL_TOKEN', None), + 'key': os.getenv('CONSUL_KEY', None), + } + + def __getattr__(self, name: str) -> Union[dict, None]: + """ + Get an environment variable as an attribute. + + :param name: The name of the environment variable. + :return: The value of the environment variable, or None if it does not exist. + """ + return os.environ.get(name) diff --git a/application/utils/loaders/consul_loader.py b/application/utils/loaders/consul_loader.py index 555c62686ea9a006847679df9721942834fcea32..cea5dbe651edfe036237ca455642a6821767ba19 100644 --- a/application/utils/loaders/consul_loader.py +++ b/application/utils/loaders/consul_loader.py @@ -10,28 +10,27 @@ # @Description : """ -import os -from typing import NamedTuple, Union +from typing import Union import requests from dynaconf.base import LazySettings from dynaconf.utils.parse_conf import parse_conf_data +from application.common import EnvVarHelper from application.lib import ConsulConfig IDENTIFIER = "consul_loader" -def get_env_vars() -> dict: - return { - 'host': os.environ.get('CONSUL_HOST', 'localhost'), - 'port': os.environ.get('CONSUL_PORT', 8500), - 'dc': os.environ.get('CONSUL_DC', 'dc1'), - 'token': os.getenv('CONSUL_TOKEN'), - } - - def parse_config(data: dict, key: str, obj: LazySettings) -> dict: + """ + Parses the configuration data. + + :param data: The configuration data. + :param key: The key to be parsed in the configuration. + :param obj: The LazySettings object. + :return: A dictionary with parsed configuration data. + """ if key is not None: data = data[key] @@ -48,10 +47,25 @@ def load( key: str = None, validate=False, ) -> Union[bool, None]: - env_vars = get_env_vars() - consul_key = os.getenv('CONSUL_KEY') - - if consul_key is None: + """ + Loads the configuration data. + + :param obj: The LazySettings object. + :param env: The environment. + :param silent: If True, suppress exceptions. + :param key: The key to be loaded in the configuration. + :param validate: If True, validate the configuration. + :return: Boolean indicating success, or None in case of failure. + """ + # 实例化对象 + env_helper = EnvVarHelper() + # 从环境变量获取 consul 参数 + env_vars = env_helper.consul_vars + # 获取 consul key + consul_key = env_vars.pop('key', None) + + # If the key is None or empty, return None + if not consul_key: return client = ConsulConfig(**env_vars) @@ -61,7 +75,8 @@ def load( except requests.exceptions.ConnectionError: return except Exception as e: - raise RuntimeError(f'Unknown error: {e}') + # Include original exception in the traceback + raise RuntimeError(f'Unknown error: {e}') from e if env is None: return