tegwick
f5619e1a52
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: sonnet Assistant-Process: 295952@bnt-lap001 Assistant-Session: e93f64ad-516c-46eb-9666-aad8d300c477
83 lines
3.5 KiB
Python
83 lines
3.5 KiB
Python
"""The flow graph of a runbook-pack/v0.2: stages, steps, alternatives, gates, canonical order."""
|
|
from __future__ import annotations
|
|
|
|
|
|
class Flow:
|
|
def __init__(self, data: dict):
|
|
self.data = data
|
|
self.stages: dict[str, dict] = {}
|
|
self.steps: dict[str, dict] = {}
|
|
self.gates: dict[str, dict] = {}
|
|
self.alts: dict[str, str] = {} # alternative id -> parent step id
|
|
self.parent: dict[str, str | None] = {} # flow id -> enclosing stage id
|
|
self.order: list[str] = [] # canonical order of steps and gates
|
|
self.datapoints = {d["id"]: d for d in data.get("datapoints", [])}
|
|
self.actors = {a["id"]: a for a in data.get("actors", [])}
|
|
self._walk(data["stages"], None)
|
|
|
|
def _walk(self, stages: list[dict], parent: str | None) -> None:
|
|
for st in stages:
|
|
self.stages[st["id"]] = st
|
|
self.parent[st["id"]] = parent
|
|
for s in st.get("steps", []):
|
|
self.steps[s["id"]] = s
|
|
self.parent[s["id"]] = st["id"]
|
|
self.order.append(s["id"])
|
|
for a in s.get("one_of", []):
|
|
self.alts[a["id"]] = s["id"]
|
|
self._walk(st.get("stages", []), st["id"])
|
|
for g in st.get("gates", []):
|
|
self.gates[g["id"]] = g
|
|
self.parent[g["id"]] = st["id"]
|
|
self.order.append(g["id"])
|
|
|
|
# -- structure ------------------------------------------------------------------------
|
|
def kind(self, id_: str) -> str:
|
|
return "stage" if id_ in self.stages else "step" if id_ in self.steps else "gate" if id_ in self.gates else "?"
|
|
|
|
def node(self, id_: str) -> dict:
|
|
return self.stages.get(id_) or self.steps.get(id_) or self.gates[id_]
|
|
|
|
def resolve(self, ref: str) -> str | None:
|
|
if ref.startswith("gate:"):
|
|
return ref[5:] if ref[5:] in self.gates else None
|
|
return ref if ref in self.stages or ref in self.steps else None
|
|
|
|
def ancestors(self, id_: str) -> list[str]:
|
|
out, p = [], self.parent.get(id_)
|
|
while p:
|
|
out.append(p)
|
|
p = self.parent.get(p)
|
|
return out
|
|
|
|
def members(self, stage_id: str, include_gates: bool = True) -> list[str]:
|
|
"""Steps (and gates) inside a stage, recursively."""
|
|
out = []
|
|
st = self.stages[stage_id]
|
|
out += [s["id"] for s in st.get("steps", [])]
|
|
for child in st.get("stages", []):
|
|
out += self.members(child["id"], include_gates)
|
|
if include_gates:
|
|
out += [g["id"] for g in st.get("gates", [])]
|
|
return out
|
|
|
|
def requires(self, id_: str) -> list[str]:
|
|
"""Own requires plus those of every enclosing stage, as resolved flow ids."""
|
|
refs = list(self.node(id_).get("requires", []))
|
|
for a in self.ancestors(id_):
|
|
refs += self.stages[a].get("requires", [])
|
|
return [r for r in (self.resolve(x) for x in refs) if r]
|
|
|
|
|
|
def pack_idempotency(data: dict) -> str:
|
|
"""Derived characteristic of the whole runbook: idempotent, non-idempotent, conditional or unknown."""
|
|
fl = Flow(data)
|
|
vals = []
|
|
for s in fl.steps.values():
|
|
parts = s.get("one_of") or [s]
|
|
vals += [p.get("idempotent", s.get("idempotent", "unknown")) for p in parts]
|
|
if not vals or "unknown" in vals:
|
|
return "unknown"
|
|
if "non-idempotent" in vals:
|
|
return "non-idempotent"
|
|
return "conditional" if "conditional" in vals else "idempotent"
|