Update project

application/common/__init__.py

@@ -11,4 +11,5 @@
 """
 
 from .config import ConfigHelper
+from .env import EnvVarHelper
 from .file import FileHelper

application/common/env.py

+#!/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)

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