Auto-update: Thu Aug 8 18:27:32 PDT 2024
This commit is contained in:
sanj
parent
1b4b242ac8
commit
15af9e8a6b
1 changed files with 48 additions and 73 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue