runbook-tutorials/rtut/project.py

"""Projects: runbooks that share an environment, with layered attributes, derived state and templates."""
from __future__ import annotations

import re
from pathlib import Path

import yaml

from . import maturity as _m  # noqa: F401  (kept importable for callers)
from . import run as R
from .attrs import Doc, Registry, check_value, fold, normalize, set_layers
from .pack import digest

PROJECT_STATES = ("planned", "applying", "applied", "stale")


# ---- validation --------------------------------------------------------------------------------------

def validate(reg: Registry, doc: Doc) -> list[str]:
    """Structural errors plus references that must resolve in the registry."""
    if not doc.ok:
        return list(doc.errors)
    d, errs = doc.data, []
    for s in d.get("attribute_sets", []):
        try:
            set_layers(reg, s)
        except ValueError as e:
            errs.append(str(e))
    targets = d.get("targets", {})
    for tid, t in targets.items():
        for s in t.get("attribute_sets", []):
            try:
                set_layers(reg, s)
            except ValueError as e:
                errs.append(f"target {tid}: {e}")
    packs = [m["pack"] for m in d["runbooks"]]
    keys = [(m["pack"], m.get("target")) for m in d["runbooks"]]
    if len(keys) != len(set(keys)):
        errs.append("a runbook appears twice for the same target")
    for m in d["runbooks"]:
        if m["pack"] not in reg.packs:
            errs.append(f"runbook {m['pack']} not found")
        elif not reg.packs[m["pack"]].ok:
            errs.append(f"runbook {m['pack']} is invalid")
        if m.get("target") and m["target"] not in targets:
            errs.append(f"runbook {m['pack']}: unknown target {m['target']}")
        for a in m.get("after", []):
            if a not in packs:
                errs.append(f"runbook {m['pack']}: after names {a}, which is not a member")
    # acyclic member order
    deps = {m["pack"]: set(m.get("after", [])) for m in d["runbooks"]}
    state: dict[str, int] = {}

    def visit(n: str, path: list[str]) -> None:
        if state.get(n) == 1:
            errs.append("member order cycle: " + " -> ".join(path[path.index(n):] + [n])); return
        if state.get(n) == 2 or n not in deps:
            return
        state[n] = 1
        for x in sorted(deps[n]):
            visit(x, path + [n])
        state[n] = 2

    for n in sorted(deps):
        visit(n, [])
    return sorted(set(errs))


# ---- attribute resolution --------------------------------------------------------------------------

def _member(d: dict, pack_id: str, target: str | None = None) -> dict:
    """The member for this runbook and target; a member without a target matches target=None."""
    for m in d["runbooks"]:
        if m["pack"] == pack_id and m.get("target") == target:
            return m
    for m in d["runbooks"]:                               # lenient: a runbook named without its target
        if m["pack"] == pack_id and target is None:
            return m
    raise KeyError(pack_id)


def layers_for(reg: Registry, d: dict, member: dict) -> list:
    layers: list = []
    for s in d.get("attribute_sets", []):
        layers += set_layers(reg, s)
    layers.append(("project", normalize(d.get("attributes"))))
    t = d.get("targets", {}).get(member.get("target") or "", None)
    if t:
        for s in t.get("attribute_sets", []):
            layers += set_layers(reg, s)
        layers.append((f"target:{member['target']}", normalize(t.get("attributes"))))
    layers.append((f"member:{member['pack']}", normalize(member.get("attributes"))))
    return layers


def member_attributes(reg: Registry, d: dict, member: dict) -> dict:
    """What the member runbook's inputs look like in this project.

    {'rows': [{id, label, value, state, requirement, source, modified, error?}], 'unused': [names]}
    state: nil (no value), set, err (invalid for the runbook). requirement: required unless the attribute
    says optional or the runbook itself has a default.
    """
    pack = reg.packs[member["pack"]]
    folded = fold(layers_for(reg, d, member))
    defs = [x for x in (pack.data.get("datapoints") or pack.data.get("parameters") or []) if not x.get("produced_by")]
    rows = []
    for x in defs:
        a = folded.get(x["id"], {"value": None, "requirement": None, "source": None, "modified": False})
        value = a["value"]
        row = {"id": x["id"], "label": x.get("label", x["id"]), "value": value, "source": a["source"],
               "modified": a["modified"], "requirement": a["requirement"] or ("optional" if "default" in x else "required")}
        if value is None:
            row["state"] = "nil"
            if "default" in x:
                row["default"] = x["default"]
        else:
            err = check_value(pack, x["id"], value)
            row["state"] = "err" if err else "set"
            if err:
                row["error"] = err
        rows.append(row)
    known = {x["id"] for x in defs}
    return {"rows": rows, "unused": sorted(n for n in folded if n not in known)}


def missing_required(reg: Registry, d: dict) -> list[dict]:
    """Attributes some member still needs: required, no value, no runbook default."""
    out = []
    for m in d["runbooks"]:
        if m["pack"] not in reg.packs:
            continue
        for r in member_attributes(reg, d, m)["rows"]:
            if r["state"] == "nil" and r["requirement"] == "required" and "default" not in r:
                out.append({"pack": m["pack"], "attribute": r["id"], "label": r["label"]})
    return out


def member_values(reg: Registry, d: dict, member: dict) -> dict[str, str]:
    """Values to hand to a run: every set attribute (invalid ones are left out and reported as err)."""
    return {r["id"]: (str(r["value"]).lower() if isinstance(r["value"], bool) else str(r["value"]))
            for r in member_attributes(reg, d, member)["rows"] if r["state"] == "set"}


# ---- state ---------------------------------------------------------------------------------------------

def _latest_run(pack, project_id: str, target: str | None):
    best = None
    for rid in R.list_runs(pack):
        r = R.load(pack, rid)
        ctx = r.data.get("context", {})
        if ctx.get("project") == project_id and ctx.get("target") == target:
            if best is None or r.data["created"] >= best.data["created"]:
                best = r
    return best


def member_status(reg: Registry, d: dict, member: dict) -> dict:
    """not-applied | applying | applied | stale | failed, from the latest run at the current pack revision."""
    pack = reg.packs[member["pack"]]
    r = _latest_run(pack, d["id"], member.get("target"))
    if r is None:
        return {"pack": member["pack"], "target": member.get("target"), "status": "not-applied", "run": None}
    eff = R.effective_outcome(r)
    current = r.data.get("pack_digest") == digest(pack)
    if eff == "succeeded":
        status = "applied" if current else "stale"
    elif eff in ("failed", "abandoned"):
        status = "failed"
    else:
        status = "applying"
    return {"pack": member["pack"], "target": member.get("target"), "status": status, "run": r.data["run_id"],
            "deviations": len(r.data.get("deviations", [])), "current_revision": current}


def state(reg: Registry, doc: Doc) -> dict:
    """Project state: planned | applying | applied | stale, with per-member status and what can start next."""
    d = doc.data
    members = [member_status(reg, d, m) for m in d["runbooks"]]
    st = [m["status"] for m in members]
    if "stale" in st:
        project = "stale"
    elif all(x == "applied" for x in st):
        project = "applied"
    elif all(x == "not-applied" for x in st):
        project = "planned"
    else:
        project = "applying"
    applied = {m["pack"] for m in members if m["status"] == "applied"}
    nxt = [m["pack"] for m, spec in zip(members, d["runbooks"])
           if m["status"] in ("not-applied", "failed", "stale") and set(spec.get("after", [])) <= applied]
    return {"state": project, "members": members, "next": nxt, "needs": missing_required(reg, d)}


# ---- runs from a project ------------------------------------------------------------------------------------

def start_run(reg: Registry, doc: Doc, pack_id: str, operator: str, target: str | None = None, mode: str = "manual",
              values: dict | None = None, extra_context: dict | None = None):
    """Create a run of a member runbook with the project's attributes as defaults and the project as context.

    `values` replaces the current attributes (a project run passes the snapshot it took at its start);
    `extra_context` adds to the run's context (for example the project run it belongs to)."""
    d = doc.data
    if d["spec"] != "project/v0.1":
        raise R.RunError("a template is not run; instantiate it into a project first")
    errs = validate(reg, doc)
    if errs:
        raise R.RunError(f"project {doc.id} is invalid: {errs[0]}")
    member = _member(d, pack_id, target)
    pack = reg.packs[pack_id]
    ctx = {"project": doc.id, "environment": d["environment"], "target": member.get("target"), **(extra_context or {})}
    values = dict(values) if values is not None else member_values(reg, d, member)
    if pack.spec == "runbook-pack/v0.2":
        from . import run2
        return run2.create(pack, operator, values, mode=mode, context=ctx)
    if mode != "manual":
        raise R.RunError("v0.1 packs run in manual mode only")
    return R.create(pack, operator, values, ctx)


# ---- templates -----------------------------------------------------------------------------------------------

def instantiate(reg: Registry, template_id: str, project_id: str, environment: str,
                values: dict[str, str] | None = None, title: str | None = None, root: Path | None = None,
                description: str = "", goal: str = "") -> Path:
    """Write projects/<id>/project.yaml from a template. Required attributes may stay unset; `state()` lists them.

    The project has its own name, description and goal. Nothing of the template's own title, goal or description
    is copied: the project only records which template it came from.
    """
    t = reg.templates.get(template_id)
    if t is None or not t.ok:
        raise ValueError(f"no valid template {template_id}")
    if not re.fullmatch(r"[a-z0-9]+([.-][a-z0-9]+)*", project_id):
        raise ValueError("project id must be lower-case letters, digits, dots and hyphens")
    if environment not in R.ENVIRONMENTS:
        raise ValueError(f"environment must be one of {', '.join(R.ENVIRONMENTS)}")
    from .pack import SECRET_RE
    for label, text in (("title", title or ""), ("description", description), ("goal", goal)):
        if SECRET_RE.search(text):
            raise ValueError(f"the {label} looks like a secret; projects are never secret")
    d = {k: v for k, v in t.data.items() if k not in ("spec", "id", "title", "goal", "description")}
    attrs = normalize(d.get("attributes"))
    for k, v in (values or {}).items():
        attrs.setdefault(k, {"value": None, "requirement": None})
        attrs[k]["value"] = v
    out_attrs = {}
    for k, a in attrs.items():
        out_attrs[k] = a["value"] if a.get("value") is not None and not a.get("requirement") and not a.get("note") else \
            {kk: vv for kk, vv in a.items() if vv is not None}
    if len((title or project_id).strip()) < 3:
        raise ValueError("the project name needs at least 3 characters")
    project = {"spec": "project/v0.1", "id": project_id, "title": (title or project_id).strip()}
    if description.strip():
        project["description"] = description.strip()
    if goal.strip():
        project["goal"] = goal.strip()
    project.update({k: v for k, v in d.items() if k not in ("attributes", "environment")})
    project.update({"environment": environment, "template": template_id})
    if out_attrs:
        project["attributes"] = out_attrs
    home = Path(root) if root else t.path.parent.parent
    target = home / "projects" / project_id / "project.yaml"
    if target.exists():
        raise ValueError(f"{target} already exists")
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(yaml.safe_dump(project, sort_keys=False))
    return target


# ---- editing -----------------------------------------------------------------------------------------------------

def set_attributes(reg: Registry, doc: Doc, values: dict[str, str]) -> Path:
    """Give attributes values in the project file itself. Refuses values a consuming runbook rejects."""
    d = doc.data
    if d["spec"] != "project/v0.1":
        raise ValueError("only a project, not a template, takes values this way")
    known: dict[str, list] = {}
    for m in d["runbooks"]:
        pack = reg.packs.get(m["pack"])
        if pack is None:
            continue
        for x in (pack.data.get("datapoints") or pack.data.get("parameters") or []):
            known.setdefault(x["id"], []).append(pack)
    attrs = dict(d.get("attributes") or {})
    for name, value in values.items():
        value = str(value).strip()
        if not value:
            continue
        if not re.fullmatch(r"[a-z][a-z0-9_]*", name):
            raise ValueError(f"bad attribute name {name}")
        from .pack import SECRET_RE
        if SECRET_RE.search(value):
            raise ValueError(f"{name}: value looks like a secret; attributes are never secret")
        for pack in known.get(name, []):
            err = check_value(pack, name, value)
            if err:
                raise ValueError(f"{name}: {err} (for {pack.id})")
        if isinstance(attrs.get(name), dict):
            attrs[name] = {**attrs[name], "value": value}
        else:
            attrs[name] = value
    text = yaml.safe_load(doc.path.read_text())
    text["attributes"] = attrs
    doc.path.write_text(yaml.safe_dump(text, sort_keys=False))
    return doc.path


def update_project(doc: Doc, title: str, description: str = "", goal: str = "") -> Path:
    """Rename a project and edit its description and goal, in its own file."""
    if doc.data.get("spec") != "project/v0.1":
        raise ValueError("only a project can be edited this way")
    from .pack import SECRET_RE
    title = title.strip()
    if len(title) < 3:
        raise ValueError("the project name needs at least 3 characters")
    for label, text in (("name", title), ("description", description), ("goal", goal)):
        if SECRET_RE.search(text):
            raise ValueError(f"the {label} looks like a secret; projects are never secret")
    data = yaml.safe_load(doc.path.read_text())
    data["title"] = title
    for key, text in (("description", description), ("goal", goal)):
        if text.strip():
            data[key] = text.strip()
        else:
            data.pop(key, None)
    ordered = {k: data[k] for k in ("spec", "id", "title", "description", "goal") if k in data}
    ordered.update({k: v for k, v in data.items() if k not in ordered})
    doc.path.write_text(yaml.safe_dump(ordered, sort_keys=False))
    return doc.path