-
Notifications
You must be signed in to change notification settings - Fork 11
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
feat: refactor config module #120
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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) |
This file was deleted.
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. though |
||
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 | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. inline imports are not encouraged. |
||
|
||
Logger.error(message=f"Config file '{self.filename}' not found.") | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The error, though look right, may not be sufficient as it could also be the path that can be the issue |
||
|
||
def merge(self, other: ConfigFile) -> None: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. As a best practice, |
||
""" | ||
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]: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Can we rename this method for better readability? |
||
""" | ||
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]: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Is there no out of box function that dataclass or python has around it? |
||
""" | ||
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]: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It is not clear why a config file has or knows about "env" variable. Something about the abstraction does not looks right. |
||
""" | ||
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: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Recommend to rename the function for better readability. The confusion arises from what "process" means here? |
||
""" | ||
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 |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Comments like this are not good, see How to write better code comments, a reference mentioned in our handbook
Please audit rest of the changes in the PR with this point of view and address them as well.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why is this class method and not static method (unlike most of our other classes)?