homepage.veen.world/app/utils/configuration_resolver.py

135 lines
5.7 KiB
Python
Raw Normal View History

2025-01-10 13:56:37 +01:00
from pprint import pprint
2025-01-09 13:53:56 +01:00
class ConfigurationResolver:
"""
A class to resolve `link` entries in a nested configuration structure.
Supports navigation through dictionaries, lists, and `children`.
2025-01-09 13:53:56 +01:00
"""
def __init__(self, config):
self.config = config
def resolve_links(self):
"""
Resolves all `link` entries in the configuration.
"""
self._recursive_resolve(self.config, self.config)
def __load_children(self,path):
"""
Check if explicitly children should be loaded and not parent
"""
return path.split('.').pop() == "children"
def _replace_in_dict_by_dict(self, dict_origine, old_key, new_dict):
if old_key in dict_origine:
# Entferne den alten Key
old_value = dict_origine.pop(old_key)
# Füge die neuen Key-Value-Paare hinzu
dict_origine.update(new_dict)
def _replace_in_list_by_list(self, list_origine, old_element, new_elements):
index = list_origine.index(old_element)
list_origine[index:index+1] = new_elements
def _replace_element_in_list(self, list_origine, old_element, new_element):
index = list_origine.index(old_element)
list_origine[index] = new_element
2025-01-09 13:53:56 +01:00
def _recursive_resolve(self, current_config, root_config):
"""
Recursively resolves `link` entries in the configuration.
"""
if isinstance(current_config, dict):
for key, value in list(current_config.items()):
if key == "children":
for item in value:
if "link" in item:
loaded_link = self._find_entry(root_config, item['link'].lower(), False)
if isinstance(loaded_link, list):
self._replace_in_list_by_list(value,item,loaded_link)
else:
2025-01-15 02:08:49 +01:00
self._replace_element_in_list(value,item,loaded_link)
else:
self._recursive_resolve(value, root_config)
elif key == "link":
2025-01-09 13:56:29 +01:00
try:
2025-01-15 02:08:49 +01:00
loaded = self._find_entry(root_config, value.lower(), True)
if isinstance(loaded, list) and len(loaded) > 2:
loaded = self._find_entry(root_config, value.lower(), False)
current_config.clear()
current_config.update(loaded)
2025-01-10 13:56:37 +01:00
except Exception as e:
2025-01-09 13:56:29 +01:00
raise ValueError(
f"Error resolving link '{value}': {str(e)}. "
f"Current path: {key}, Current config: {current_config}" + (f", Loaded: {loaded}" if 'loaded' in locals() or 'loaded' in globals() else "")
2025-01-09 13:56:29 +01:00
)
2025-01-09 13:53:56 +01:00
else:
self._recursive_resolve(value, root_config)
elif isinstance(current_config, list):
for item in current_config:
self._recursive_resolve(item, root_config)
def _get_children(self,current):
if isinstance(current, dict) and ("children" in current and current["children"]):
current = current["children"]
2025-01-10 13:56:37 +01:00
return current
def _find_by_name(self,current, part):
return next(
(item for item in current if isinstance(item, dict) and item.get("name", "").lower() == part),
None
)
def _find_entry(self, config, path, children):
2025-01-09 13:53:56 +01:00
"""
Finds an entry in the configuration by a dot-separated path.
Supports both dictionaries and lists with `children` navigation.
2025-01-09 13:53:56 +01:00
"""
parts = path.split('.')
current = config
for part in parts:
if isinstance(current, list):
# If children explicit declared just load children
if part != "children":
# Look for a matching name in the list
found = self._find_by_name(current,part)
if found:
current = found
print(
f"Matching entry for '{part}' in list. Path so far: {' > '.join(parts[:parts.index(part)+1])}. "
f"Current list: {current}"
)
else:
raise ValueError(
f"No matching entry for '{part}' in list. Path so far: {' > '.join(parts[:parts.index(part)+1])}. "
f"Current list: {current}"
)
2025-01-09 13:53:56 +01:00
elif isinstance(current, dict):
# Case-insensitive dictionary lookup
key = next((k for k in current if k.lower() == part), None)
if key is None:
current = self._find_by_name(current["children"],part)
2025-01-10 13:56:37 +01:00
if not current:
raise KeyError(
f"Key '{part}' not found in dictionary. Path so far: {' > '.join(parts[:parts.index(part)+1])}. "
f"Current dictionary: {current}"
)
else:
current = current[key]
2025-01-09 13:53:56 +01:00
else:
2025-01-09 13:56:29 +01:00
raise ValueError(
f"Invalid path segment '{part}'. Current type: {type(current)}. "
f"Path so far: {' > '.join(parts[:parts.index(part)+1])}"
)
if children:
current = self._get_children(current)
2025-01-09 13:53:56 +01:00
return current
def get_config(self):
"""
Returns the resolved configuration.
"""
2025-01-09 13:56:29 +01:00
return self.config