diff --git a/src/apps/backend/modules/config/config_service.py b/src/apps/backend/modules/config/config_service.py index bb2023ed..7b035374 100644 --- a/src/apps/backend/modules/config/config_service.py +++ b/src/apps/backend/modules/config/config_service.py @@ -1,22 +1,34 @@ -from typing import Generic,TypeVar, cast +from typing import Generic, Optional, cast + from modules.common.types import ErrorCode +from modules.config.internals.config_manager import ConfigManager +from modules.config.types import T from modules.error.custom_errors import MissingKeyError -from modules.config.internals.config import Config -T = TypeVar('T') class ConfigService(Generic[T]): - @staticmethod - def load_config() -> None: - Config.load_config() + config_manager: ConfigManager = ConfigManager() + + @classmethod + def load_config(cls) -> None: + """ + Load the configuration files + """ + cls.config_manager.load_config() @classmethod - def get_value(cls,key: str) -> T: - value = Config.get(key) + def get_value(cls, key: str, default: Optional[T] = None) -> T: + """ + Get the value of the key from the configuration + """ + value: Optional[T] = cls.config_manager.get(key, default=default) if value is None: raise MissingKeyError(missing_key=key, error_code=ErrorCode.MISSING_KEY) - return cast(T,value) + return cast(T, value) - @staticmethod - def has_value(key: str) -> bool: - return Config.has(key) + @classmethod + def has_value(cls, key: str) -> bool: + """ + Check if the key exists in the configuration + """ + return cls.config_manager.has(key) diff --git a/src/apps/backend/modules/config/internals/config.py b/src/apps/backend/modules/config/internals/config.py deleted file mode 100644 index ccdfdf87..00000000 --- a/src/apps/backend/modules/config/internals/config.py +++ /dev/null @@ -1,156 +0,0 @@ -import os -from pathlib import Path -from typing import Any, Optional -import yaml - - -class Config: - _config_dict: dict[str, Any] - _config_path: Path - - @staticmethod - def load_config() -> None: - Config._intialize_config() - Config._process_custom_environment_variables() - print(yaml.dump(Config._config_dict)) - - @staticmethod - def get(key: str, default: Optional[Any] = None) -> Any: - keys = key.split(".") - value = Config._config_dict - try: - for k in keys: - value = value[k] - return value - except (KeyError, TypeError): - return default - - @staticmethod - def has(key: str) -> bool: - keys = key.split(".") - value = Config._config_dict - try: - for k in keys: - value = value[k] - return True - except (KeyError, TypeError): - return False - - @staticmethod - def _read(filename: str) -> dict[str, Any]: - file_path = Config._config_path / filename - yaml_content: dict[str, Any] = {} - try: - with open(file_path, "r", encoding="utf-8") as file: - yaml_content = yaml.safe_load(file) - except FileNotFoundError as e: - print(e) - return yaml_content - - @staticmethod - def _intialize_config() -> None: - Config._config_path = Config._get_base_directory(__file__) / "config" - default_content = Config._read("default.yml") - app_env = os.environ.get("APP_ENV", "development") - app_env_content = Config._read(f"{app_env}.yml") - merge_content = Config._deep_merge(default_content, app_env_content) - Config._config_dict = merge_content - - @staticmethod - def _parse_value(value:Optional[str], value_format:str)-> Any: - """ - Parse the environment variable value based on the specified format. - """ - if value is None: - return None - - parsers = { - "boolean": lambda x: x.lower() in ["true", "1"], - "number": lambda x: int(x) if x.isdigit() else float(x), - } - - parser = parsers.get(value_format) - if not parser: - raise ValueError(f"Unsupported format: {value_format}") - - try: - return parser(value) - except Exception as e: - raise ValueError(f"Error parsing value '{value}' as {value_format}: {e}") from e - - - @staticmethod - def _replace_with_env_values(data:dict[str,Any]) -> dict[str,Any]: - """ - Recursively traverse a dictionary and replace values with the corresponding environment variable values - if the value in the dictionary matches a key in the environment variables. - """ - if not isinstance(data, dict): - return data - - keys_to_delete:list = [] # Collect keys to delete - - for key, value in data.items(): - if isinstance(value, dict): - data[key] = Config._process_dict_value(value) - elif isinstance(value, str): - Config._process_str_value(data, key, value, keys_to_delete) - - Config._delete_keys(data, keys_to_delete) - return data - - @staticmethod - def _process_dict_value(value : dict[str,Any]) -> Any: - """Process a dictionary value, replacing or recursively traversing it.""" - if "__name" in value: - env_var_name = value["__name"] - env_var_value = os.getenv(env_var_name) - value_format = value.get("__format") - return Config._parse_value(env_var_value, value_format) if value_format else env_var_value - return Config._replace_with_env_values(value) - - @staticmethod - def _process_str_value(data:dict[str,Any], key:str, value:str, keys_to_delete:list) -> None: - """Process a string value, replacing it with an environment variable or marking for deletion.""" - env_value = os.getenv(value) - if env_value is None: - keys_to_delete.append(key) - else: - data[key] = env_value - - @staticmethod - def _delete_keys(data:dict[str,Any], keys_to_delete:list) -> None: - """Delete keys marked for deletion.""" - for key in keys_to_delete: - del data[key] - - @staticmethod - def _get_base_directory(current_file: str) -> Path: - starting_index = current_file.find("app") - base_directory = current_file[:starting_index+len("app")] - return Path(base_directory) - - @staticmethod - def _process_custom_environment_variables() -> None: - """ - Reads keys from custom_env_contents, maps them to environment variables, - and overrides them in the configuration if they exist. - """ - custom_env_contents = Config._read("custom-environment-variables.yml") - replaced_custom_env_contents = Config._replace_with_env_values(custom_env_contents) - Config._deep_merge(Config._config_dict, replaced_custom_env_contents) - - @staticmethod - def _deep_merge(dict1: dict[str, Any], dict2: dict[str, Any]) -> dict[str, Any]: - """ - Recursively merges dict2 into dict1. Values from dict2 will overwrite those in dict1. - If a value is a nested dictionary, it will be merged as well. - """ - for key, value in dict2.items(): - if isinstance(value, dict) and key in dict1 and isinstance(dict1[key], dict): - # If both are dictionaries, merge them recursively - dict1[key] = Config._deep_merge(dict1[key], value) - else: - # If not a dictionary, just overwrite or add the key-value pair - dict1[key] = value - return dict1 diff --git a/src/apps/backend/modules/config/internals/config_file.py b/src/apps/backend/modules/config/internals/config_file.py new file mode 100644 index 00000000..99beb6fc --- /dev/null +++ b/src/apps/backend/modules/config/internals/config_file.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any, Optional + +import yaml + + +class ConfigFile: + """ + Represents a configuration file that can be loaded, merged, and updated with environment variables. + """ + + def __init__(self, filename: str, config_path: Path): + self.filename = filename + self.config_path = config_path + self.content: dict[str, Any] = {} + + def load(self) -> None: + """ + Loads the content of the configuration file. + """ + file_path = self.config_path / self.filename + try: + with open(file_path, "r", encoding="utf-8") as file: + self.content = yaml.safe_load(file) or {} + except FileNotFoundError: + from modules.logger.logger import Logger + + Logger.error(message=f"Config file '{self.filename}' not found.") + + def merge(self, other: ConfigFile) -> None: + """ + Merges the content of another ConfigFile into this one. + """ + self.content = self._deep_merge(self.content, other.content) + + def replace_with_env_variables(self) -> None: + """ + Replaces placeholders in the configuration file with corresponding environment variables. + """ + self.content = self._replace_with_env_values(self.content) + + def get_content(self) -> dict[str, Any]: + """ + Returns the loaded content of the config file. + """ + return self.content + + @staticmethod + def _deep_merge(dict1: dict[str, Any], dict2: dict[str, Any]) -> dict[str, Any]: + """ + Deep merge two dictionaries. + """ + for key, value in dict2.items(): + if isinstance(value, dict) and key in dict1 and isinstance(dict1[key], dict): + dict1[key] = ConfigFile._deep_merge(dict1[key], value) + else: + dict1[key] = value + return dict1 + + @staticmethod + def _replace_with_env_values(data: dict[str, Any]) -> dict[str, Any]: + """ + Replace dictionary values with corresponding environment variables if defined. + """ + if not isinstance(data, dict): + return data + + keys_to_delete: list = [] + + for key, value in data.items(): + if isinstance(value, dict): + data[key] = ConfigFile._process_dict_value(value) + elif isinstance(value, str): + ConfigFile._process_str_value(data, key, value, keys_to_delete) + + for key in keys_to_delete: + del data[key] + + return data + + @staticmethod + def _process_dict_value(value: dict[str, Any]) -> Any: + """ + Process dictionary values, replacing environment variables where necessary. + """ + if "__name" in value: + env_var_name = value["__name"] + env_var_value = os.getenv(env_var_name) + value_format = value.get("__format") + return ConfigFile._parse_value(env_var_value, value_format) if value_format else env_var_value + return ConfigFile._replace_with_env_values(value) + + @staticmethod + def _process_str_value(data: dict[str, Any], key: str, value: str, keys_to_delete: list) -> None: + """ + Replace a string value with an environment variable if it exists. If not found, mark the key for deletion. + """ + env_value = os.getenv(value) + if env_value is None: + keys_to_delete.append(key) + else: + data[key] = env_value + + @staticmethod + def _parse_value(value: Optional[str], value_format: str) -> int | float | bool | None: + """ + Parse a value based on the specified format ('boolean' or 'number'). + """ + if value is None: + return None + + parsers = { + "boolean": lambda x: x.lower() in ["true", "1"], + "number": lambda x: int(x) if x.isdigit() else float(x), + } + + parser = parsers.get(value_format) + if not parser: + raise ValueError(f"Unsupported format: {value_format}") + + try: + return parser(value) + except Exception as e: + raise ValueError(f"Error parsing value '{value}' as {value_format}: {e}") from e diff --git a/src/apps/backend/modules/config/internals/config_manager.py b/src/apps/backend/modules/config/internals/config_manager.py new file mode 100644 index 00000000..a1269ff8 --- /dev/null +++ b/src/apps/backend/modules/config/internals/config_manager.py @@ -0,0 +1,68 @@ +import os +from pathlib import Path +from typing import Any, Optional + +from modules.config.internals.config_file import ConfigFile + + +class ConfigManager: + """ + Manages application configuration by loading, merging, and providing access to config values. + """ + + CONFIG_KEY_SEPARATOR: str = "." + + def __init__(self) -> None: + self.config_store: dict[str, Any] = {} + self.config_path = self._get_base_directory(__file__) / "config" + + def get(self, key: str, default: Optional[Any] = None) -> Optional[Any]: + """ + Retrieve a configuration value using dot-separated keys. + """ + value = self.config_store + for k in key.split(self.CONFIG_KEY_SEPARATOR): + if not isinstance(value, dict) or k not in value: + return default + value = value[k] + return value + + def has(self, key: str) -> bool: + """ + Check if a configuration key exists. + """ + value = self.config_store + for k in key.split(self.CONFIG_KEY_SEPARATOR): + if not isinstance(value, dict) or k not in value: + return False + value = value[k] + return True + + def load_config(self) -> None: + """ + Load and merge configuration files from the config directory. + """ + default_config = ConfigFile("default.yml", self.config_path) + default_config.load() + + app_env = os.environ.get("APP_ENV", "development") + app_env_config = ConfigFile(f"{app_env}.yml", self.config_path) + app_env_config.load() + default_config.merge(app_env_config) + + custom_env_config = ConfigFile("custom-environment-variables.yml", self.config_path) + custom_env_config.load() + custom_env_config.replace_with_env_variables() + + default_config.merge(custom_env_config) + + self.config_store = default_config.get_content() + + @staticmethod + def _get_base_directory(current_file: str) -> Path: + """ + Get the base directory of the project. + """ + starting_index = current_file.find("app") + base_directory = current_file[: starting_index + len("app")] + return Path(base_directory) diff --git a/src/apps/backend/modules/config/types.py b/src/apps/backend/modules/config/types.py index b1c0c644..84d08515 100644 --- a/src/apps/backend/modules/config/types.py +++ b/src/apps/backend/modules/config/types.py @@ -1,4 +1,7 @@ from dataclasses import dataclass +from typing import TypeVar + +T = TypeVar("T", bound=int | str | bool | list | dict) @dataclass(frozen=True) diff --git a/src/apps/backend/modules/logger/internal/loggers.py b/src/apps/backend/modules/logger/internal/loggers.py index 0e81b0d9..3f246038 100644 --- a/src/apps/backend/modules/logger/internal/loggers.py +++ b/src/apps/backend/modules/logger/internal/loggers.py @@ -11,7 +11,7 @@ class Loggers: @staticmethod def initialize_loggers() -> None: - logger_transports = ConfigService[list[str]].get_value(key='logger.transports') + logger_transports = ConfigService[list[str]].get_value(key="logger.transports") for logger_transport in logger_transports: if logger_transport == LoggerTransports.CONSOLE: Loggers._loggers.append(Loggers.__get_console_logger()) diff --git a/src/apps/backend/modules/logger/internal/papertrail_logger.py b/src/apps/backend/modules/logger/internal/papertrail_logger.py index fcafeed7..561adc00 100644 --- a/src/apps/backend/modules/logger/internal/papertrail_logger.py +++ b/src/apps/backend/modules/logger/internal/papertrail_logger.py @@ -1,8 +1,8 @@ import logging from logging.handlers import SysLogHandler -from modules.config.types import PapertrailConfig from modules.config.config_service import ConfigService +from modules.config.types import PapertrailConfig from modules.logger.internal.base_logger import BaseLogger @@ -13,8 +13,8 @@ def __init__(self) -> None: # Create a console handler and set the level to INFO logger_config = PapertrailConfig( - host=ConfigService[str].get_value(key='papertrail.host'), - port=ConfigService[int].get_value(key='papertrail.port') + host=ConfigService[str].get_value(key="papertrail.host"), + port=ConfigService[int].get_value(key="papertrail.port"), ) papertrail_handler = SysLogHandler(address=(logger_config.host, logger_config.port)) formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") diff --git a/src/apps/backend/server.py b/src/apps/backend/server.py index 2767fd89..c974d970 100644 --- a/src/apps/backend/server.py +++ b/src/apps/backend/server.py @@ -23,8 +23,10 @@ # Apply ProxyFix to interpret `X-Forwarded` headers if enabled in configuration # Visit: https://flask.palletsprojects.com/en/stable/deploying/proxy_fix/ for more information -if ConfigService.has_value("is_server_running_behind_proxy") and ConfigService[bool].get_value("is_server_running_behind_proxy"): - app.wsgi_app = ProxyFix(app.wsgi_app) # type: ignore +if ConfigService.has_value("is_server_running_behind_proxy") and ConfigService[bool].get_value( + "is_server_running_behind_proxy" +): + app.wsgi_app = ProxyFix(app.wsgi_app) # type: ignore # Register access token apis access_token_blueprint = AccessTokenRestApiServer.create()