diff --git a/application/common/__init__.py b/application/common/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..004634da5dcf20580c320129936483ea4ab00642 --- /dev/null +++ b/application/common/__init__.py @@ -0,0 +1,14 @@ +#!/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 diff --git a/application/common/config.py b/application/common/config.py new file mode 100644 index 0000000000000000000000000000000000000000..cb3b0a0a62764db24c53c1ed66d7e6b97f18b307 --- /dev/null +++ b/application/common/config.py @@ -0,0 +1,43 @@ +#!/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"" + + def __str__(self): + return f"ConfigHelper for app {self.app}" diff --git a/application/common/file.py b/application/common/file.py new file mode 100644 index 0000000000000000000000000000000000000000..2ac383e8a7592c543113ff56e1ca09ae8a0a4140 --- /dev/null +++ b/application/common/file.py @@ -0,0 +1,48 @@ +#!/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." diff --git a/application/lib/config/local.py b/application/lib/config/local.py index 0ff170244047b40d4e1372485d6d75934a742dc6..c2c81a5f01766c7d4260ec500441879e3398cfa8 100644 --- a/application/lib/config/local.py +++ b/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 diff --git a/application/lib/flask_elasticsearch/elasticsearch.py b/application/lib/flask_elasticsearch/elasticsearch.py index 88ff27162420fb1f0e7ca82e663b0709c0d19fe4..b1b74cde247817554cbd3d9b6fbe121538b9c9a7 100644 --- a/application/lib/flask_elasticsearch/elasticsearch.py +++ b/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.') diff --git a/application/utils/__init__.py b/application/utils/__init__.py index 796c9b16afcb9027da3807a11d94c0323e7a2b55..b33a1fc8872a08fff90fdd23cf1e7f0cbe556111 100644 --- a/application/utils/__init__.py +++ b/application/utils/__init__.py @@ -10,5 +10,5 @@ # @Description : """ -from .dsn import dsn, DatabaseURI +from .dsn import DatabaseURI from .elasticsearch import ElasticsearchUtils diff --git a/application/utils/dsn/dsn.py b/application/utils/dsn/dsn.py index 13cc031272f7a6e4be9c2b60c4cc3198c5b526b9..8fb56fdee24974e14c12b77baaa8457fc5afcc0f 100644 --- a/application/utils/dsn/dsn.py +++ b/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"" + return f'' def __str__(self): return self.create() diff --git a/application/utils/loaders/consul_loader.py b/application/utils/loaders/consul_loader.py index 58297fc738b1b5a78f6746ebec0626ca4ce33d96..555c62686ea9a006847679df9721942834fcea32 100644 --- a/application/utils/loaders/consul_loader.py +++ b/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, + ) diff --git a/application/utils/loaders/yaml_loader.py b/application/utils/loaders/yaml_loader.py index 7234afda059b5f3d04886e10bef6025251e1b7b4..e5d7e0a012ba74cbaabdd72c888ed12b643d4771 100644 --- a/application/utils/loaders/yaml_loader.py +++ b/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, + )