Update project

application/common/__init__.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/3 19:42
+# @File        : __init__.py
+# @Description :
+"""
+
+from .config import ConfigHelper
+from .file import FileHelper

application/common/config.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/3 19:43
+# @File        : config.py
+# @Description :
+"""
+
+
+from typing import Any
+
+from flask import Flask
+
+
+class ConfigHelper:
+    """
+    The ConfigHelper class is a utility for fetching configuration values
+    from a Flask application.
+
+    :param app: The Flask application instance from which to fetch configuration values.
+    """
+
+    def __init__(self, app: Flask):
+        self.app = app
+
+    def get_config(self, key: str) -> Any:
+        """
+        Fetch a config value based on the provided key.
+
+        :param key: The key for the config value.
+        :return: The value for the provided key.
+        """
+        return self.app.config.get(key)
+
+    def __repr__(self):
+        return f"<ConfigHelper with app {self.app}>"
+
+    def __str__(self):
+        return f"ConfigHelper for app {self.app}"

application/common/file.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/3 19:58
+# @File        : file.py
+# @Description :
+"""
+
+
+class FileHelper:
+    """
+    FileHelper is a utility class that provides file-related operations.
+    Currently, it only provides a method to get the configuration file name
+    based on the environment.
+    """
+
+    @classmethod
+    def get_filename(cls, env: str) -> str:
+        """
+        Get the configuration file name based on the environment.
+
+        :param env: a string representing the environment.
+                    It should be 'PRODUCTION' or None.
+        :return: a string representing the configuration file name.
+                 If env is 'PRODUCTION', return 'production_config.yaml'.
+                 Otherwise, return 'config.yaml'.
+        """
+        if env == 'PRODUCTION':
+            return f'{env.lower()}_config.yaml'
+        else:
+            return 'config.yaml'
+
+    def __repr__(self):
+        """
+        Return a string representing a valid Python expression that could be used
+        to recreate the FileHelper object.
+        """
+        return "FileHelper()"
+
+    def __str__(self):
+        """
+        Return a human-readable string representation of the FileHelper object.
+        """
+        return "This is a FileHelper class that helps with file related operations."

application/lib/config/local.py

@@ -23,7 +23,7 @@ class LocalConfig:
     def load(self, filename: str) -> Any:
         filepath = os.path.join(self.config_dir, filename)
         if not os.path.exists(filepath):
-            raise FileNotFoundError(f"No such file or directory: '{filepath}'")
+            raise FileNotFoundError(f'No such file or directory: {filepath}')
         with open(filepath, 'r') as file:
             cfg = yaml.safe_load(file)
         return cfg

application/lib/flask_elasticsearch/elasticsearch.py

@@ -16,6 +16,8 @@ from elasticsearch import Elasticsearch
 from flask import current_app, Flask
 from loguru import logger
 
+from application.common import ConfigHelper
+
 
 class FlaskElasticsearch:
     def __init__(self, app=None):
@@ -42,7 +44,7 @@ class FlaskElasticsearch:
         """Lazy initialization of Elasticsearch connection on first use."""
         ctx = current_app._get_current_object()
         if ctx is not None:
-            if not hasattr(ctx, "elasticsearch"):
+            if not hasattr(ctx, 'elasticsearch'):
                 cfg = self._get_config()
                 ctx.elasticsearch = Elasticsearch(**cfg)
                 if ctx.elasticsearch.ping():
@@ -57,29 +59,20 @@ class FlaskElasticsearch:
         """Retrieves Elasticsearch configuration from the current Flask application context."""
         with current_app.app_context():
             if current_app:
-                # Retrieve configuration from current_app.config and return it
-                host = current_app.config.Elasticsearch.Host if current_app.config.get(
-                    'Elasticsearch') is not None else 'localhost'
-
-                port = int(current_app.config.Elasticsearch.Port) if current_app.config.get(
-                    'Elasticsearch') is not None else 9200
-
-                user = current_app.config.Elasticsearch.User if current_app.config.get(
-                    'Elasticsearch') is not None else 'user'
-
-                password = current_app.config.Elasticsearch.Password if current_app.config.get(
-                    'Elasticsearch') is not None else 'password'
-
-                use_ssl = current_app.config.Elasticsearch.UseSsl == 'True' if current_app.config.get(
-                    'Elasticsearch') is not None else False
-
-                verify_certs = current_app.config.Elasticsearch.VerifyCerts == 'False' if current_app.config.get(
-                    'Elasticsearch') is not None else True
-
-                ca_certs = current_app.config.Elasticsearch.CaCerts if current_app.config.get(
-                    'Elasticsearch') is not None else None
-
-                es_config = dict(
+                config_helper = ConfigHelper(current_app)
+                cfg = config_helper.get_config('Elasticsearch')
+                if cfg is None:
+                    raise KeyError('Key Elasticsearch is not defined')
+
+                host = cfg.Host or 'localhost'
+                port = int(cfg.Port) or 9200
+                user = cfg.User or None
+                password = cfg.Password or None
+                use_ssl = cfg.UseSsl == 'True' or False
+                verify_certs = cfg.VerifyCerts == 'False'
+                ca_certs = cfg.CaCerts or None
+
+                options = dict(
                     hosts=[{'host': host, 'port': port}],
                     http_auth=None if not user else (user, password),
                     use_ssl=use_ssl,
@@ -87,7 +80,7 @@ class FlaskElasticsearch:
                     ca_certs=ca_certs
                 )
 
-                return es_config
+                return options
             else:
                 logger.error('Attempted to access application configuration outside of application context.')
                 raise RuntimeError('Attempted to access application configuration outside of application context.')

application/utils/__init__.py

@@ -10,5 +10,5 @@
 # @Description :
 """
 
-from .dsn import dsn, DatabaseURI
+from .dsn import DatabaseURI
 from .elasticsearch import ElasticsearchUtils

application/utils/dsn/dsn.py

@@ -60,7 +60,7 @@ class DatabaseURI:
                 return f'{self.db_type}://{self.username}:{self.password}@{self.host}:{self.port}/{self.db}'
 
     def __repr__(self):
-        return f"<DatabaseURI({self.db_type})>"
+        return f'<DatabaseURI({self.db_type})>'
 
     def __str__(self):
         return self.create()

application/utils/loaders/consul_loader.py

@@ -22,11 +22,23 @@ from application.lib import ConsulConfig
 IDENTIFIER = "consul_loader"
 
 
-class SourceMetadata(NamedTuple):
-    loader: str
-    identifier: str
-    env: str
-    merged: bool = False
+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:
+    if key is not None:
+        data = data[key]
+
+    return {
+        key: parse_conf_data(value, tomlfy=True, box_settings=obj)
+        for key, value in data.items()
+    }
 
 
 def load(
@@ -36,51 +48,34 @@ def load(
     key: str = None,
     validate=False,
 ) -> Union[bool, None]:
-    consul_host = os.environ.get('CONSUL_HOST', 'localhost')
-    consul_port = os.environ.get('CONSUL_PORT', 8500)
-    consul_dc = os.environ.get('CONSUL_DC', 'dc1')
-    consul_token = os.getenv('CONSUL_TOKEN')
+    env_vars = get_env_vars()
     consul_key = os.getenv('CONSUL_KEY')
 
-    # 没有对应环境变量，会进入下一个加载器
     if consul_key is None:
         return
 
-    # 实例 ConsulConfig
-    client = ConsulConfig(host=consul_host, port=consul_port, token=consul_token, dc=consul_dc)
-    # 捕获异常
+    client = ConsulConfig(**env_vars)
+
     try:
         data = client.get(key=consul_key)
     except requests.exceptions.ConnectionError:
-        print(2)
-        # 连接错误后，则会进入下一个加载器
         return
     except Exception as e:
-        # 发生未知错误，才会抛出异常
         raise RuntimeError(f'Unknown error: {e}')
 
-    # 基于 key 获取所需配置
-    if key is not None:
-        data = data[key]
-
     if env is None:
         return
 
     try:
-        # 获取 consul 注册中心的配置信息
-        result = {
-            key: parse_conf_data(value, tomlfy=True, box_settings=obj)
-            for key, value in data.items()
-        }
+        result = parse_config(data, key, obj)
     except Exception as e:
         if silent:
             return False
         raise e
-    else:
-        result['Consul'] = True
-        # 将 result 配置写入 dynaconf 内置配置中
-        obj.update(
-            result,
-            loader_identifier=IDENTIFIER,
-            validate=validate,
-        )
+
+    result['Consul'] = True
+    obj.update(
+        result,
+        loader_identifier=IDENTIFIER,
+        validate=validate,
+    )

application/utils/loaders/yaml_loader.py

@@ -17,11 +17,48 @@ from typing import Union
 from dynaconf.base import LazySettings
 from dynaconf.utils.parse_conf import parse_conf_data
 
+from application.common import FileHelper
 from application.lib import LocalConfig
 
 IDENTIFIER = "yaml_loader"
 
 
+def load_config(local_cfg: LocalConfig, filename: str, key: str) -> dict:
+    """
+    Load the configuration from a file.
+
+    :param local_cfg: a LocalConfig object for loading the config.
+    :param filename: a string representing the name of the config file.
+    :param key: a string representing the key to extract from the config. If it's None, return the whole config.
+    :return: a dict representing the loaded config.
+    """
+    if key is None:
+        return local_cfg.load(filename)
+    else:
+        return local_cfg.load(filename)[key]
+
+
+def parse_config(cfg: dict, env: str, obj: LazySettings) -> dict:
+    """
+    Parse the configuration based on the environment.
+
+    :param cfg: a dict representing the loaded config.
+    :param env: a string representing the environment. It should be 'PRODUCTION' or None.
+    :param obj: a LazySettings object that the config will be applied to.
+    :return: a dict representing the parsed config.
+    """
+    if cfg.get(env) is None:
+        return {
+            key: parse_conf_data(value, tomlfy=True, box_settings=obj)
+            for key, value in cfg.items()
+        }
+    else:
+        return {
+            key: parse_conf_data(value, tomlfy=True, box_settings=obj)
+            for key, value in cfg[env].items()
+        }
+
+
 def load(
     obj: LazySettings,
     env: str = None,
@@ -29,46 +66,35 @@ def load(
     key: str = None,
     validate=False,
 ) -> Union[bool, None]:
-    # 判断是否已经加载 consul 配置，如果加载，则不再加载本地配置
+    """
+    Load and apply the configuration to a LazySettings object.
+
+    :param obj: a LazySettings object that the config will be applied to.
+    :param env: a string representing the environment. It should be 'PRODUCTION' or None.
+    :param silent: a bool indicating whether to suppress the KeyError when the key is not in the config.
+    :param key: a string representing the key to extract from the config. If it's None, apply the whole config.
+    :param validate: a bool indicating whether to validate the config after loading.
+    :return: None if the config is successfully applied, False if the key is not in the config and silent is True.
+    """
     if obj.get('Consul', False):
         return
-    # 获取本地配置文件目录
+
     config_dir = os.path.join(str(Path(__file__).parent.parent.parent), 'config')
-    # 实例化 LocalConfig 对象
     local_cfg = LocalConfig(config_dir)
 
-    # 判断当前是否为生产环境，基于不同环境读取本地不同配置文件
-    if env == 'PRODUCTION':
-        filename = f'{env.lower()}_config.yaml'
-    else:
-        filename = 'config.yaml'
-
-    # 基于 key 获取所需配置
-    if key is None:
-        cfg = local_cfg.load(filename)
-    else:
-        cfg = local_cfg.load(filename)[key]
+    filename = FileHelper.get_filename(env)
+    cfg = load_config(local_cfg, filename, key)
 
     try:
-        # 将 cfg 配置写入 dynaconf 内置配置中
-        if cfg.get(env) is None:
-            result = {
-                key: parse_conf_data(value, tomlfy=True, box_settings=obj)
-                for key, value in cfg.items()
-            }
-        else:
-            result = {
-                key: parse_conf_data(value, tomlfy=True, box_settings=obj)
-                for key, value in cfg[env].items()
-            }
-    except Exception as e:
+        result = parse_config(cfg, env, obj)
+    except KeyError as e:
         if silent:
             return False
         raise e
-    else:
-        if result:
-            obj.update(
-                result,
-                loader_identifier=IDENTIFIER,
-                validate=validate,
-            )
+
+    if result:
+        obj.update(
+            result,
+            loader_identifier=IDENTIFIER,
+            validate=validate,
+        )