"""What a key rename has to move, read where each kind of thing keeps names. Only the fields classified as holding message names, never a text search: a flow from before ids keeps its key as its id, in its directory and its history rows; a deleted flow that had the same key keeps its history; and payloads, logs and git history keep whatever they say. """ import json from collections.abc import Iterable from pathlib import Path from typing import Any from sqlmodel import Session, col, select from fluksio.core.config import settings from fluksio.core.db import engine from fluksio.flow.dashboards import DASHBOARD_BINDINGS, DASHBOARD_DIR from fluksio.flow.messages import flow_of from fluksio.flow.namespace import FLOW_FIELDS, NAME, OUTPUTS, Rename, names, touches from fluksio.flow.queue import WorkQueue from fluksio.flow.rename import DEDUP, SQL_NAMES from fluksio.flow.state import StateBackend from fluksio.models import ( AlertSent, EngineEvent, FlowRun, Run, RunArtifact, RunMetric, RunNode, ) #: The history tables by name, as `SQL_NAMES` has them. _TABLES = {model.__tablename__: model for model in (RunMetric, RunArtifact, RunNode)} def _documents(root: Path) -> Iterable[tuple[Path, tuple[tuple[str, ...], ...]]]: for path in sorted(root.glob("*/flow*.json")): # `_devices/flow.json` is a device, not a flow. if not path.parent.name.startswith(("_", ".")): yield path, FLOW_FIELDS for path in sorted((root / DASHBOARD_DIR).glob("*/dashboard*.json")): yield path, DASHBOARD_BINDINGS def leftovers( key: str, root: Path, flow_ids: Iterable[str], state: StateBackend | None = None, queue: WorkQueue | None = None, ) -> list[str]: """Every classified field that still names something of the flow key *key*: document keys and bindings under *root*, the history of the flows *flow_ids*, the alert ledger and channels, live state and queued work.""" def named(name: str) -> bool: return "." in name and flow_of(name) == key found = [] for path, paths in _documents(root): rel = path.relative_to(root) data = json.loads(path.read_text()) if paths is FLOW_FIELDS and data.get("name") == key: found.append(f"{rel}: its key") found += [f"{rel}: {name}" for name in names(data, paths) if named(name)] runs = select(Run.id).where(col(Run.flow_id).in_([*flow_ids])) with Session(engine) as session: for (table, column), holds in SQL_NAMES.items(): if holds == DEDUP: # The tail is a node, or a flow key on its own. found += [ f"{table}.{column}: {one}" for one in session.exec(select(AlertSent.key)).all() if flow_of(one.partition(":")[2]) == key ] continue model = _TABLES[table] values = session.exec( select(getattr(model, column)).where(col(model.run_id).in_(runs)) ).all() if holds == NAME: found += [f"{table}.{column}: {v}" for v in values if named(v)] elif holds == OUTPUTS: found += [ f"{table}.{column}: {name}" for text in values for name in json.loads(text or "{}") if named(name) ] try: alerts = json.loads(settings.ALERTS_FILE.read_text()) except FileNotFoundError: alerts = {} found += [ f"alerts.json: {channel['config']['message']}" for channel in alerts.get("channels", []) if channel["kind"] == "dashboard" and named(channel["config"]["message"]) ] if state is not None: found += [one for one in state.keys() if touches(key, one)] found += [f"names: {one}" for one in state.message_names() if named(one)] if queue is not None: probe = Rename(key, f"{key}_probe") found += [ f"queued: {one.node}" for one in queue.live_items() if probe.item(one) ] markers = queue.count(probe)["done_markers"] found += [f"done markers: {markers}"] if markers else [] return found #: Columns that hold ids, which a rename never changes. _IDENTITIES: tuple[Any, ...] = ( Run.id, Run.flow_id, Run.for_node, RunNode.node, RunMetric.node, RunArtifact.node, FlowRun.id, FlowRun.flow_id, ) def stability(root: Path) -> dict[str, Any]: """What a rename leaves exactly as it was: the store's directories, every file in it but the documents, and the id columns of the history (`EngineEvent` without the audit rows a rename adds).""" documents = {path for path, _ in _documents(root)} files = { str(path.relative_to(root)): path.read_bytes() for path in sorted(root.rglob("*")) if path.is_file() and ".git" not in path.parts and path not in documents } directories = sorted( str(path.relative_to(root)) for path in root.rglob("*") if path.is_dir() and ".git" not in path.parts ) with Session(engine) as session: ids = { str(column): sorted(map(str, session.exec(select(column)).all())) for column in _IDENTITIES } ids["engine_event"] = sorted( map( str, session.exec( select(EngineEvent.flow_id, EngineEvent.node).where( col(EngineEvent.type) != "audit" ) ).all(), ) ) return {"files": files, "directories": directories, "ids": ids}