runbook-tutorials/rtut/attrs.py

"""Attribute sets: named, composable, non-secret attribute collections with layered resolution.

An attribute has a *requirement* (required | optional), declared, and a *state*, derived with the same
words as run datapoints: nil (no value), set, err (invalid for a consuming runbook). `modified` marks a
value that a later layer replaced. Every value keeps its source, so a wrong value is traceable.
"""
from __future__ import annotations

import json
from dataclasses import dataclass, field
from pathlib import Path

import yaml
from jsonschema import Draft202012Validator

from . import pack as P
from .pack import SECRET_RE

SCHEMA_DIR = Path(__file__).resolve().parent.parent / "schema"


def _schema_errors(data: dict, schema_file: str) -> list[str]:
    v = Draft202012Validator(json.loads((SCHEMA_DIR / schema_file).read_text()))
    return [f"{'/'.join(map(str, e.absolute_path)) or '<root>'}: {e.message}"
            for e in sorted(v.iter_errors(data), key=lambda e: list(map(str, e.absolute_path)))]


def normalize(attrs: dict | None) -> dict[str, dict]:
    """scalar or {value, requirement, note} -> {value, requirement, note}"""
    out = {}
    for k, v in (attrs or {}).items():
        out[k] = dict(v) if isinstance(v, dict) else {"value": v}
        out[k].setdefault("value", None)
        out[k].setdefault("requirement", None)
    return out


@dataclass
class Doc:
    path: Path
    data: dict
    errors: list[str] = field(default_factory=list)

    @property
    def id(self) -> str:
        return str(self.data.get("id", self.path.stem))

    @property
    def ok(self) -> bool:
        return not self.errors


def _load(path: Path, schema_file: str) -> Doc:
    try:
        data = yaml.safe_load(Path(path).read_text())
    except yaml.YAMLError as e:
        return Doc(Path(path), {}, [f"invalid YAML: {e}"])
    if not isinstance(data, dict):
        return Doc(Path(path), {}, ["must be a mapping"])
    errs = _schema_errors(data, schema_file)
    if not errs and SECRET_RE.search(json.dumps(data)):
        errs.append("contains secret-looking marker")
    return Doc(Path(path), data, errs)


def load_set(path: Path) -> Doc:
    return _load(path, "attribute-set.schema.json")


class Registry:
    """Everything found under one or more repository roots."""

    def __init__(self, roots: list[Path], extra_sets: dict[str, Doc] | None = None):
        self.roots = [Path(r) for r in roots]
        self.sets: dict[str, Doc] = {}
        self.projects: dict[str, Doc] = {}
        self.templates: dict[str, Doc] = {}
        self.packs: dict[str, P.Pack] = {}
        for r in self.roots:
            for f in sorted(r.glob("attribute-sets/*.yaml")):
                d = load_set(f); self.sets[d.id] = d
            for f in sorted(r.glob("projects/*/project.yaml")):
                d = _load(f, "project.schema.json"); self.projects[d.id] = d
            for f in sorted(r.glob("project-templates/*.yaml")):
                d = _load(f, "project.schema.json"); self.templates[d.id] = d
            for f in P.discover(r):
                pk = P.load(f); self.packs[pk.id] = pk
        for sid, doc in (extra_sets or {}).items():          # app-level sets (e.g. the user): repository sets win
            self.sets.setdefault(sid, doc)


def set_layers(reg: Registry, set_id: str, _seen: tuple = ()) -> list[tuple[str, dict]]:
    """Layers of an attribute set, its `extends` first. Raises on unknown sets and cycles."""
    if set_id in _seen:
        raise ValueError("attribute set cycle: " + " -> ".join(_seen + (set_id,)))
    doc = reg.sets.get(set_id)
    if doc is None:
        raise ValueError(f"unknown attribute set {set_id}")
    if not doc.ok:
        raise ValueError(f"attribute set {set_id} is invalid: {doc.errors[0]}")
    layers: list[tuple[str, dict]] = []
    for base in doc.data.get("extends", []):
        layers += set_layers(reg, base, _seen + (set_id,))
    layers.append((f"set:{set_id}", normalize(doc.data["attributes"])))
    return layers


def fold(layers: list[tuple[str, dict]]) -> dict[str, dict]:
    """Later layers win. Keeps provenance and marks a value replaced by a later layer as modified."""
    out: dict[str, dict] = {}
    for source, attrs in layers:
        for name, a in attrs.items():
            cur = out.setdefault(name, {"value": None, "requirement": None, "source": None, "modified": False,
                                        "history": [], "note": ""})
            if a.get("requirement"):
                cur["requirement"] = a["requirement"]
            if a.get("note"):
                cur["note"] = a["note"]
            if a.get("value") is not None:
                if cur["value"] is not None and cur["value"] != a["value"]:
                    cur["modified"] = True
                cur["value"] = a["value"]
                cur["source"] = source
                cur["history"].append({"source": source, "value": a["value"]})
    for cur in out.values():
        cur["state"] = "set" if cur["value"] is not None else "nil"
    return out


def check_value(pack: P.Pack, name: str, value) -> str | None:
    """None if the value is valid for the pack's datapoint/parameter `name`, else the reason."""
    defs = {d["id"]: d for d in (pack.data.get("datapoints") or pack.data.get("parameters") or [])}
    d = defs.get(name)
    if d is None:
        return None
    v = str(value).lower() if isinstance(value, bool) else str(value)
    import re
    if d["type"] == "int" and not re.fullmatch(r"-?\d+", v):
        return "not an integer"
    if d["type"] == "bool" and v not in ("true", "false"):
        return "must be true or false"
    try:
        P.render("{{%s}}" % name, {name: v}, pack)
    except (ValueError, KeyError) as e:
        return str(e)
    return None