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 / "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


# ---- managing attribute sets ---------------------------------------------------------------------------------------

import re as _re

SET_ID = _re.compile(r"[a-z0-9]+([.-][a-z0-9]+)*")
ATTR_NAME = _re.compile(r"[a-z][a-z0-9_]*")


def used_by(reg: Registry, set_id: str) -> dict[str, list[str]]:
    """Who reuses an attribute set: projects and templates (also through targets), and sets that extend it."""
    out = {"projects": [], "templates": [], "sets": []}
    for kind, docs in (("projects", reg.projects), ("templates", reg.templates)):
        for did, d in docs.items():
            refs = list(d.data.get("attribute_sets", []))
            for t in d.data.get("targets", {}).values():
                refs += t.get("attribute_sets", [])
            if set_id in refs:
                out[kind].append(did)
    for sid, d in reg.sets.items():
        if sid != set_id and set_id in d.data.get("extends", []):
            out["sets"].append(sid)
    return out


def _would_cycle(reg: Registry, set_id: str, extends: list[str]) -> bool:
    def reaches(start: str, seen: set) -> bool:
        if start == set_id:
            return True
        if start in seen or start not in reg.sets:
            return False
        seen.add(start)
        return any(reaches(x, seen) for x in reg.sets[start].data.get("extends", []))
    return any(reaches(x, set()) for x in extends)


def save_set(reg: Registry, root: Path, set_id: str, title: str, extends: list[str],
             attributes: dict[str, dict], existing_path: Path | None = None) -> Path:
    """Write an attribute set file. `attributes` maps name -> {value, requirement, note} (any part may be empty).

    Validates the id, title, names, secrets, that extended sets exist and that nothing becomes circular.
    """
    if not SET_ID.fullmatch(set_id or ""):
        raise ValueError("the id is lower-case letters, digits, dots and hyphens")
    if len((title or "").strip()) < 3:
        raise ValueError("the title needs at least 3 characters")
    extends = [x for x in extends if x]
    for x in extends:
        if x == set_id:
            raise ValueError("a set cannot extend itself")
        if x not in reg.sets:
            raise ValueError(f"unknown attribute set {x} in extends")
    if _would_cycle(reg, set_id, extends):
        raise ValueError("that extends list would make the sets circular")
    out: dict = {}
    for name, a in attributes.items():
        if not ATTR_NAME.fullmatch(name):
            raise ValueError(f"bad attribute name {name!r}: lower-case letters, digits and underscores, starting with a letter")
        value = a.get("value")
        value = None if value is None or str(value).strip() == "" else str(value).strip()
        req = a.get("requirement") or None
        if req not in (None, "required", "optional"):
            raise ValueError(f"{name}: requirement is required or optional")
        note = (a.get("note") or "").strip() or None
        if value is None and req is None and note is None:
            raise ValueError(f"{name}: give a value, a requirement or a note")
        for text in (value or "", note or ""):
            if SECRET_RE.search(text):
                raise ValueError(f"{name}: looks like a secret; attribute sets are never secret")
        entry = {k: v for k, v in (("value", value), ("requirement", req), ("note", note)) if v is not None}
        out[name] = entry["value"] if list(entry) == ["value"] else entry
    if SECRET_RE.search(title):
        raise ValueError("the title looks like a secret")
    doc: dict = {"spec": "attribute-set/v0.1", "id": set_id, "title": title.strip()}
    if extends:
        doc["extends"] = extends
    doc["attributes"] = out
    errs = _schema_errors(doc, "attribute-set.schema.json")
    if errs:
        raise ValueError(errs[0])
    path = existing_path or (Path(root) / "attribute-sets" / f"{set_id}.yaml")
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(yaml.safe_dump(doc, sort_keys=False))
    return path