From 15af9e8a6b6fa069eaab2dd56eb62afab6e58fee Mon Sep 17 00:00:00 2001 From: sanj <67624670+iodrift@users.noreply.github.com> Date: Thu, 8 Aug 2024 18:27:32 -0700 Subject: [PATCH] Auto-update: Thu Aug 8 18:27:32 PDT 2024 --- sijapi/classes.py | 121 ++++++++++++++++++---------------------------- 1 file changed, 48 insertions(+), 73 deletions(-) diff --git a/sijapi/classes.py b/sijapi/classes.py index 1d50ecb..1c274f1 100644 --- a/sijapi/classes.py +++ b/sijapi/classes.py @@ -83,23 +83,21 @@ TS_ID = os.environ.get('TS_ID') T = TypeVar('T', bound='Configuration') - class Configuration(BaseModel): HOME: Path = Path.home() _dir_config: Optional['Configuration'] = None - + @classmethod def load(cls, yaml_path: Union[str, Path], secrets_path: Optional[Union[str, Path]] = None, dir_config: Optional['Configuration'] = None) -> 'Configuration': yaml_path = cls._resolve_path(yaml_path, 'config') if secrets_path: secrets_path = cls._resolve_path(secrets_path, 'config') - + try: with yaml_path.open('r') as file: config_data = yaml.safe_load(file) - + debug(f"Loaded configuration data from {yaml_path}") - secrets_data = {} if secrets_path: with secrets_path.open('r') as file: secrets_data = yaml.safe_load(file) @@ -108,43 +106,39 @@ class Configuration(BaseModel): for item in config_data: if isinstance(item, dict): item.update(secrets_data) - elif isinstance(config_data, dict): + else: config_data.update(secrets_data) - if isinstance(config_data, list): config_data = {"configurations": config_data} - if not isinstance(config_data, dict) or config_data.get('HOME') is None: - config_data = config_data if isinstance(config_data, dict) else {} + if config_data.get('HOME') is None: config_data['HOME'] = str(Path.home()) - debug(f"HOME was not in config, set to default: {config_data['HOME']}") - + warn(f"HOME was None in config, set to default: {config_data['HOME']}") + load_dotenv() instance = cls.create_dynamic_model(**config_data) instance._dir_config = dir_config or instance - resolved_data = instance.resolve_placeholders(config_data, secrets_data) + resolved_data = instance.resolve_placeholders(config_data) instance = cls.create_dynamic_model(**resolved_data) instance._dir_config = dir_config or instance return instance - + except Exception as e: err(f"Error loading configuration: {str(e)}") raise - - + @classmethod def _resolve_path(cls, path: Union[str, Path], default_dir: str) -> Path: - base_path = Path(__file__).parent.parent + base_path = Path(__file__).parent.parent # This will be two levels up from this file path = Path(path) if not path.suffix: path = base_path / 'sijapi' / default_dir / f"{path.name}.yaml" elif not path.is_absolute(): path = base_path / path return path - - - def resolve_placeholders(self, data: Any, secrets_data: Dict[str, Any]) -> Any: + + def resolve_placeholders(self, data: Any) -> Any: if isinstance(data, dict): - resolved_data = {k: self.resolve_placeholders(v, secrets_data) for k, v in data.items()} + resolved_data = {k: self.resolve_placeholders(v) for k, v in data.items()} home_dir = Path(resolved_data.get('HOME', self.HOME)).expanduser() base_dir = Path(__file__).parent.parent data_dir = base_dir / "data" @@ -153,17 +147,16 @@ class Configuration(BaseModel): resolved_data['DATA'] = str(data_dir) return resolved_data elif isinstance(data, list): - return [self.resolve_placeholders(v, secrets_data) for v in data] + return [self.resolve_placeholders(v) for v in data] elif isinstance(data, str): - return self.resolve_string_placeholders(data, secrets_data, Path(self.HOME)) + return self.resolve_string_placeholders(data) else: return data - - + def resolve_string_placeholders(self, value: str) -> Any: pattern = r'\{\{\s*([^}]+)\s*\}\}' matches = re.findall(pattern, value) - + for match in matches: parts = match.split('.') if len(parts) == 1: # Internal reference @@ -172,21 +165,16 @@ class Configuration(BaseModel): replacement = getattr(self, parts[1], str(Path.home() / parts[1].lower())) elif len(parts) == 2 and parts[0] == 'ENV': replacement = os.getenv(parts[1], '') - elif len(parts) == 2 and parts[0] == 'SECRET': - replacement = getattr(self, parts[1].strip(), '') - if not replacement: - warn(f"Secret '{parts[1].strip()}' not found in secrets file") else: replacement = value - + value = value.replace('{{' + match + '}}', str(replacement)) - + # Convert to Path if it looks like a file path if isinstance(value, str) and (value.startswith(('/', '~')) or (':' in value and value[1] == ':')): return Path(value).expanduser() return value - @classmethod def create_dynamic_model(cls, **data): for key, value in data.items(): @@ -194,47 +182,45 @@ class Configuration(BaseModel): data[key] = cls.create_dynamic_model(**value) elif isinstance(value, list) and all(isinstance(item, dict) for item in value): data[key] = [cls.create_dynamic_model(**item) for item in value] - + DynamicModel = create_model( f'Dynamic{cls.__name__}', __base__=cls, **{k: (Any, v) for k, v in data.items()} ) return DynamicModel(**data) - - + class Config: extra = "allow" arbitrary_types_allowed = True - - + class DirConfig(BaseModel): HOME: Path = Path.home() - + @classmethod def load(cls, yaml_path: Union[str, Path]) -> 'DirConfig': yaml_path = cls._resolve_path(yaml_path, 'config') - + try: with yaml_path.open('r') as file: config_data = yaml.safe_load(file) - + print(f"Loaded configuration data from {yaml_path}") - + # Ensure HOME is set if 'HOME' not in config_data: config_data['HOME'] = str(Path.home()) print(f"HOME was not in config, set to default: {config_data['HOME']}") - + instance = cls.create_dynamic_model(**config_data) - resolved_data = instance.resolve_placeholders(config_data, {}) + resolved_data = instance.resolve_placeholders(config_data) return cls.create_dynamic_model(**resolved_data) - + except Exception as e: print(f"Error loading configuration: {str(e)}") raise - + @classmethod def _resolve_path(cls, path: Union[str, Path], default_dir: str) -> Path: base_path = Path(__file__).parent.parent @@ -244,10 +230,10 @@ class DirConfig(BaseModel): elif not path.is_absolute(): path = base_path / path return path - - def resolve_placeholders(self, data: Any, secrets_data: Dict[str, Any]) -> Any: + + def resolve_placeholders(self, data: Any) -> Any: if isinstance(data, dict): - resolved_data = {k: self.resolve_placeholders(v, secrets_data) for k, v in data.items()} + resolved_data = {k: self.resolve_placeholders(v) for k, v in data.items()} home_dir = Path(resolved_data.get('HOME', self.HOME)).expanduser() base_dir = Path(__file__).parent.parent data_dir = base_dir / "data" @@ -256,39 +242,28 @@ class DirConfig(BaseModel): resolved_data['DATA'] = str(data_dir) return resolved_data elif isinstance(data, list): - return [self.resolve_placeholders(v, secrets_data) for v in data] + return [self.resolve_placeholders(v) for v in data] elif isinstance(data, str): - return self.resolve_string_placeholders(data, secrets_data) + return self.resolve_string_placeholders(data) else: return data - - def resolve_string_placeholders(self, value: str, secrets_data: Dict[str, Any]) -> Any: + + def resolve_string_placeholders(self, value: str) -> Path: pattern = r'\{\{\s*([^}]+)\s*\}\}' matches = re.findall(pattern, value) - + for match in matches: - parts = match.split('.') - if len(parts) == 1: # Internal reference - replacement = getattr(self, parts[0], str(Path(self.HOME) / parts[0].lower())) - elif len(parts) == 2 and parts[0] == 'Dir': - replacement = getattr(self, parts[1], str(Path(self.HOME) / parts[1].lower())) - elif len(parts) == 2 and parts[0] == 'ENV': - replacement = os.getenv(parts[1], '') - elif len(parts) == 2 and parts[0] == 'SECRET': - replacement = secrets_data.get(parts[1].strip(), '') - if not replacement: - warn(f"Secret '{parts[1].strip()}' not found in secrets file") + if match == 'HOME': + replacement = str(self.HOME) + elif hasattr(self, match): + replacement = str(getattr(self, match)) else: replacement = value - - value = value.replace('{{' + match + '}}', str(replacement)) - - # Convert to Path if it looks like a file path - if isinstance(value, str) and (value.startswith(('/', '~')) or (':' in value and value[1] == ':')): - return Path(value).expanduser() - return value - + value = value.replace('{{' + match + '}}', replacement) + + return Path(value).expanduser() + @classmethod def create_dynamic_model(cls, **data): DynamicModel = create_model( @@ -297,7 +272,7 @@ class DirConfig(BaseModel): **{k: (Path, v) for k, v in data.items()} ) return DynamicModel(**data) - + class Config: arbitrary_types_allowed = True