computer-playbook/utils/roles/applications/config.py

import re
from collections.abc import Mapping
from pathlib import Path

from ansible.errors import AnsibleFilterError, AnsibleUndefinedVariable
from ansible.template import AnsibleUndefined

from utils.cache.yaml import load_yaml_any
from utils.roles.mapping import ROLE_FILE_META_SECRETS


class AppConfigKeyError(AnsibleFilterError, ValueError):
    """
    Raised when a required application config key is missing (strict mode).
    Compatible with Ansible error handling and Python ValueError.
    """


class ConfigEntryNotSetError(AppConfigKeyError):
    """
    Raised when a config entry is defined in schema but not set in application.
    """


# Exception: the sentinel must not be False - genuine False leaves would be coerced to the default.
_MISSING = object()


def get(
    applications,
    application_id,
    config_path,
    strict=True,
    default=None,
    skip_missing_app=False,
):
    schema_path = str(Path("roles") / application_id / ROLE_FILE_META_SECRETS)

    def schema_defines(path):
        if not Path(schema_path).is_file():
            return False
        schema = load_yaml_any(schema_path, default_if_missing={}) or {}
        node = schema
        for part in path.split("."):
            key_match = re.match(r"^([a-zA-Z0-9_-]+)", part)
            if not key_match:
                return False
            k = key_match.group(1)
            if isinstance(node, dict) and k in node:
                node = node[k]
            else:
                return False
        return True

    def access(obj, key, path_trace):
        m = re.match(r"^([a-zA-Z0-9_-]+)(?:\[(\d+)\])?$", key)
        if not m:
            raise AppConfigKeyError(
                f"Invalid key format in config_path: '{key}'\n"
                f"Full path so far: {'.'.join(path_trace)}\n"
                f"application_id: {application_id}\n"
                f"config_path: {config_path}"
            )
        k, idx = m.group(1), m.group(2)

        if isinstance(obj, (AnsibleUndefined, AnsibleUndefinedVariable)):
            if not strict:
                return _MISSING
            raise AppConfigKeyError(
                f"Key '{k}' is undefined at '{'.'.join(path_trace)}'\n"
                f"  actual type: {type(obj).__name__}\n"
                f"  repr(obj): {obj!r}\n"
                f"  repr(applications): {applications!r}\n"
                f"application_id: {application_id}\n"
                f"config_path: {config_path}"
            )

        if isinstance(obj, Mapping):
            if k not in obj:
                if not strict:
                    return _MISSING
                trace_path = ".".join(path_trace[1:])
                if schema_defines(trace_path):
                    raise ConfigEntryNotSetError(
                        f"Config entry '{trace_path}' is defined in schema at '{schema_path}' but not set in application '{application_id}'. "
                        f"Present keys at this level: {sorted(obj.keys()) if isinstance(obj, Mapping) else type(obj).__name__}."
                    )
                raise AppConfigKeyError(
                    f"Key '{k}' not found in dict at '{key}'\n"
                    f"Full path so far: {'.'.join(path_trace)}\n"
                    f"Current object: {obj!r}\n"
                    f"application_id: {application_id}\n"
                    f"config_path: {config_path}"
                )
            obj = obj[k]
        else:
            if not strict:
                return _MISSING
            raise AppConfigKeyError(
                f"Expected dict for '{k}', got {type(obj).__name__} at '{key}'\n"
                f"Full path so far: {'.'.join(path_trace)}\n"
                f"Current object: {obj!r}\n"
                f"application_id: {application_id}\n"
                f"config_path: {config_path}"
            )

        if idx is not None:
            if not isinstance(obj, list):
                if not strict:
                    return _MISSING
                raise AppConfigKeyError(
                    f"Expected list for '{k}[{idx}]', got {type(obj).__name__}\n"
                    f"Full path so far: {'.'.join(path_trace)}\n"
                    f"Current object: {obj!r}\n"
                    f"application_id: {application_id}\n"
                    f"config_path: {config_path}"
                )
            i = int(idx)
            if i >= len(obj):
                if not strict:
                    return _MISSING
                raise AppConfigKeyError(
                    f"Index {i} out of range for list at '{k}'\n"
                    f"Full path so far: {'.'.join(path_trace)}\n"
                    f"Current object: {obj!r}\n"
                    f"application_id: {application_id}\n"
                    f"config_path: {config_path}"
                )
            obj = obj[i]
        return obj

    path_trace = [f"applications[{application_id!r}]"]
    try:
        obj = applications[application_id]
    except KeyError:
        if skip_missing_app:
            return default if default is not None else False
        raise AppConfigKeyError(
            f"Application ID '{application_id}' not found in applications dict.\n"
            f"path_trace: {path_trace}\n"
            f"applications keys: {list(applications.keys())}\n"
            f"config_path: {config_path}"
        ) from None

    for part in config_path.split("."):
        path_trace.append(part)
        obj = access(obj, part, path_trace)
        if obj is _MISSING:
            return default if default is not None else False
    return obj