alakazam.py

@@ -24,6 +24,7 @@ from uptime_kuma_api import UptimeKumaApi, MonitorType
 from time import sleep
 
 COMBINE_PATH = os.path.dirname(os.path.realpath(__file__)) + "/combine.yml"
+NON_APP_KEYS = {'CONFIG-SETS', 'GLOBALS'}
 # INSTANCE_CONFIGS: dict: contains all app organized by recipe names and instance domains
 # The structure of the dictionary is as follows:
 # {
@@ -194,6 +195,34 @@ def merge_all_group_configs(root_path: Path) -> Dict[str, Dict[str, Any]]:
     return merged_configs
 
 
+
+def get_config_set_app_configs(
+    active_config_sets: Dict[str, Any], config_sets: Dict[str, Any]
+) -> Dict[str, Any]:
+    """
+    Merges per-app configurations from all active config-sets into a single dictionary.
+    Config-sets disabled with False are skipped; unknown set names log a warning.
+    When multiple active sets define config for the same app, later sets take precedence.
+
+    Args:
+        active_config_sets (dict): Merged CONFIG-SETS dict (set name -> True or False).
+        config_sets (dict): Full config-set definitions loaded from config-sets.yml.
+
+    Returns:
+        dict: A merged dictionary of per-app configurations from all active config-sets.
+    """
+    config_set_app_configs: Dict[str, Any] = {}
+    for config_set_name, config_set_enabled in active_config_sets.items():
+        if config_set_enabled:
+            if config_set_name not in config_sets:
+                logging.warning(
+                    f"Config-set '{config_set_name}' is enabled but not defined in config-sets.yml. Skipping."
+                )
+            else:
+                config_set_app_configs = merge_dict(config_set_app_configs, config_sets[config_set_name] or {})
+    return config_set_app_configs
+
+
 def substitute_jinja_variable(jinja_dict, subs_dict) -> None:
     """
     This function recursively traverses the given jinja_dict and wherever it finds a jinja template variable, it replaces it with the corresponding value from the subs_dict.
@@ -211,39 +240,51 @@ def substitute_jinja_variable(jinja_dict, subs_dict) -> None:
                 jinja_dict[key] = template.render(subs_dict)
 
 
-def merge_instance_configs(group_config: Dict[str, Any], instance_domain: str, instance_config: Dict[str, Any]) -> Dict[str, Any]:
+def merge_instance_configs(group_config: Dict[str, Any], instance_domain: str, instance_config: Dict[str, Any], config_sets: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
     """
     Merge instance-specific configurations with group configurations to produce a consolidated configuration dictionary.
     This considers domain-specific overrides and merges them appropriately with the defaults specified at the group level.
+    Active config-sets (from config-sets.yml) are merged between group and instance configs (group < config-set < instance).
 
     Args:
         group_config (dict): A dictionary containing the group configurations from higher hierarchy levels.
         instance_domain (str): The domain associated with the instance, used for subdomain mappings.
         instance_config (dict): The configuration dictionary specific to an instance.
+        config_sets (dict): Named config-set definitions. Defaults to empty dict.
 
     Returns:
         dict: A dictionary with merged configuration for the instance, including domain mappings and server settings.
     """
+    if config_sets is None:
+        config_sets = {}
+    global_vars = merge_dict(
+        group_config.get('GLOBALS') or {},
+        instance_config.get('GLOBALS') or {},
+    )
+    server = global_vars.get('server') or instance_domain
+    active_config_sets = merge_dict(
+        group_config.get('CONFIG-SETS') or {},
+        instance_config.get('CONFIG-SETS') or {},
+    )
+    config_set_app_configs = get_config_set_app_configs(active_config_sets, config_sets)
+    instance_apps = {k: v for k, v in instance_config.items() if k not in NON_APP_KEYS}
+    # Warn about config-set entries that target apps not listed in the instance config.
+    for config_set_app in set(config_set_app_configs) - set(instance_apps):
+        logging.warning(
+            f"Config-set specifies config for '{config_set_app}' but '{config_set_app}' is not listed "
+            f"in the instance config for '{instance_domain}'. Skipping."
+        )
     merged_config = {}
-    for app, app_config in instance_config.items():
-        if app_config and group_config.get(app):
-            merged_config[app] = merge_dict(group_config[app], app_config)
-        elif app_config:
-            merged_config[app] = app_config
-        elif group_config.get(app):
-            merged_config[app] = group_config[app].copy()
-        else:
-            merged_config[app] = {}
+    for app in instance_apps:
+        group_app_config = group_config.get(app) or {}
+        config_set_app_config = config_set_app_configs.get(app) or {}
+        instance_app_config = instance_apps[app] or {}
+        # Priority: group < config-set < instance
+        merged_config[app] = merge_dict(merge_dict(group_app_config, config_set_app_config), instance_app_config)
         merged_config[app]['app_domain'] = map_subdomain(app, instance_domain, merged_config[app])
-        if not ((server:= get_value(merged_config, 'GLOBALS', 'server')) or (server:= get_value(group_config, 'GLOBALS', 'server'))):
-            server = instance_domain
         if not merged_config[app].get('server'):
             merged_config[app]['server'] = server
-    if not (global_vars:= merged_config.get('GLOBALS')):
-        global_vars = group_config.get('GLOBALS')
     substitute_jinja_variable(merged_config, global_vars)
-    if merged_config.get('GLOBALS'):
-        merged_config.pop('GLOBALS')
     return merged_config
 
 
@@ -268,7 +309,7 @@ def map_subdomain(recipe: str, instance_domain: str, app_config: Dict[str, Any])
     return domain
 
 
-def get_merged_instance_configs(config_path: Path, group_configs: Dict[str, Any], exclude_paths: List[str]) -> Dict[str, Dict[str, Any]]:
+def get_merged_instance_configs(config_path: Path, group_configs: Dict[str, Any], exclude_paths: List[str], config_sets: Optional[Dict[str, Any]] = None) -> Dict[str, Dict[str, Any]]:
     """
     Traverse a directory structure to read and merge all YAML configuration files for each instance.
     This function supports a hierarchical configuration approach by aggregating configurations across directories.
@@ -276,15 +317,18 @@ def get_merged_instance_configs(config_path: Path, group_configs: Dict[str, Any]
     Args:
         config_path (str): The path to the directory containing instance-specific YAML files.
         group_configs (dict): A dictionary containing group configurations from higher hierarchy levels.
+        config_sets (dict): Named config-set definitions passed through to merge_instance_configs. Defaults to empty dict.
 
     Returns:
         dict: A dictionary with domains as keys and their respective merged configurations as values.
     """
+    if config_sets is None:
+        config_sets = {}
     if config_path.is_file():
         parent_path = os.path.dirname(config_path)
         instance_config = read_config(str(config_path))
         domain = config_path.name.removesuffix('.yml').removesuffix('.yaml')
-        merged_config = merge_instance_configs(group_configs[parent_path], domain, instance_config)
+        merged_config = merge_instance_configs(group_configs[parent_path], domain, instance_config, config_sets)
         return {domain: merged_config}
     instances = {}
     for root, _, files in os.walk(config_path):
@@ -298,7 +342,7 @@ def get_merged_instance_configs(config_path: Path, group_configs: Dict[str, Any]
             if re.match(pattern, file):
                 instance_config = read_config(f'{root}/{file}')
                 domain = file.removesuffix('.yml').removesuffix('.yaml')
-                merged_config = merge_instance_configs(group_configs[root], domain, instance_config)
+                merged_config = merge_instance_configs(group_configs[root], domain, instance_config, config_sets)
                 instances[domain] = merged_config
     return instances
 
@@ -988,12 +1032,13 @@ def cli(loglevel: str, group_path: str, exclude:Tuple[str]) -> None:
         logging.error(f"{_root_path} does not contain {_group_path}?")
         exit(1)
     all_group_configs = merge_all_group_configs(_root_path)
+    config_sets = read_config(str(_root_path / "config-sets.yml"))
     exclude_paths = list(map(lambda p: str(Path(p).absolute()), exclude))
     if ABRA_DIR.is_relative_to(_root_path) and str(ABRA_DIR) not in exclude_paths:
         exclude_paths.append(str(ABRA_DIR))
-    instance_configs = get_merged_instance_configs(_group_path, all_group_configs, exclude_paths)
+    instance_configs = get_merged_instance_configs(_group_path, all_group_configs, exclude_paths, config_sets)
     INSTANCE_CONFIGS = merge_connection_configs(instance_configs)
-    all_configs = get_merged_instance_configs(_root_path, all_group_configs, exclude_paths)
+    all_configs = get_merged_instance_configs(_root_path, all_group_configs, exclude_paths, config_sets)
     ALL_CONFIGS = merge_connection_configs(all_configs)
     if loglevel:
         numeric_level = getattr(logging, loglevel.upper(), None)

examples/config-sets.yml

@@ -0,0 +1,29 @@
+# config-sets.yml
+# Place this file at your alakazam root path.
+# Each top-level key is a named config-set containing per-app configurations.
+# Activate a set in any instance .yml or alaka.yml:
+#
+#   CONFIG-SETS:
+#     bbb: true
+#     calendar: true
+#
+# Disable a set inherited from a group alaka.yml by setting it to false:
+#
+#   CONFIG-SETS:
+#     bbb: false
+#
+
+bbb:
+  authentik:
+    env:
+      APPLICATIONS:
+        BBB: https://nextcloud.example.com/apps/bbb
+  nextcloud:
+    initial-hooks:
+      - app install_bbb
+
+calendar:
+  authentik:
+    env:
+      APPLICATIONS:
+        Calendar: https://nextcloud.example.com/apps/calendar/