"""Flows on disk, versioned with git. Every flow is a directory: ``flow.json`` for the structure, ``nodes/*.py`` for node logic. The whole tree is a git repository and each saved change is a commit, so a flow's history is readable with ordinary git tooling and two flows never collide in one file. Editing is separated from running. The editor writes ``flow.draft.json`` and ``nodes.draft/*.py``; the engine reads only the published ``flow.json`` and ``nodes/*.py``. Publishing promotes the draft over the published files and removes it, so a flow directory without draft files is simply one with nothing unpublished — which is what every flow written before this existed looks like. """ from __future__ import annotations import json import logging import shutil import subprocess import threading from pathlib import Path from fluksio.flow.schemas import FlowDef logger = logging.getLogger(__name__) #: Where the sources of nodes shared between flows live. LIB_DIR = "_lib" DEFAULT_SOURCE = '''"""A new node. Return a dict keyed by your output ports.""" def process(): return {} ''' class FlowNotFound(KeyError): def __init__(self, name: str) -> None: super().__init__(name) self.name = name def __str__(self) -> str: return f"No flow named '{self.name}'" class FlowExists(ValueError): def __init__(self, name: str) -> None: super().__init__(name) self.name = name def __str__(self) -> str: return f"There is already a flow named '{self.name}'" class LibNotFound(KeyError): def __init__(self, name: str) -> None: super().__init__(name) self.name = name def __str__(self) -> str: return f"No shared node named '{self.name}'" class LibExists(ValueError): def __init__(self, name: str) -> None: super().__init__(name) self.name = name def __str__(self) -> str: return f"There is already a shared node named '{self.name}'" class StaleVersion(ValueError): """Someone else saved this flow since the version being written was read.""" def __init__(self, name: str, current: int) -> None: super().__init__(name) self.name = name self.current = current def __str__(self) -> str: return f"Flow '{self.name}' has changed since you loaded it" def _same_content(left: FlowDef, right: FlowDef) -> bool: """Equal but for the version counter, which the server owns.""" return left.model_copy(update={"version": 0}) == right.model_copy( update={"version": 0} ) class FlowStore: """Reads and writes flows, committing every change.""" #: One warning per process when there is no git to commit with. _warned_no_git = False #: Commits between housekeeping runs. Auto-gc is off (see `_git`), so #: nothing packs unless it is asked for; a commit counter is the trigger #: because the process that needs it never restarts — a seeding session #: commits thousands of times inside one long-lived container, which is how #: a store reached 1898 commits, 5863 loose objects and no pack at all. _GC_EVERY = 500 def __init__(self, root: Path) -> None: self.root = root self.root.mkdir(parents=True, exist_ok=True) # Draft writes are check-and-set, so two clients saving at once must not # interleave between reading the current version and writing the next. self._write_lock = threading.Lock() #: Bumped on every commit, so something derived from every flow at once #: can tell whether it is still current without reading them all again. #: In memory rather than `head()`, which is a git subprocess per call — #: it therefore counts from zero per process and misses an edit made on #: disk behind the API, which no writer here does. self.revision = 0 if not (self.root / ".git").exists(): self._git("init", "-q") self._commit("Initialise flow store", allow_empty=True) # ------------------------------------------------------------------------- # git # ------------------------------------------------------------------------- def _git(self, *args: str) -> subprocess.CompletedProcess[str]: try: return subprocess.run( # `gc.auto=0` on every invocation, because `git commit` otherwise # forks a background `gc --auto` and returns without waiting for # it. `subprocess.run` reaps the commit, not its grandchild, so # the daemon is reparented onto PID 1 — the FastAPI process, # which reaps nothing, and the zombies accumulate one per save. # Packing is asked for explicitly in `_commit` instead. ["git", "-C", str(self.root), "-c", "gc.auto=0", *args], capture_output=True, text=True, check=False, ) except FileNotFoundError: # No git on this machine — a `pip install` on a locked-down host is # where this happens. The flows are files either way, which is what # the store is; what is lost is their history, so say it once and # carry on rather than refusing to start. if not FlowStore._warned_no_git: FlowStore._warned_no_git = True logger.warning( "git is not installed, so flow changes are not versioned. " "Install it to get a commit per save." ) return subprocess.CompletedProcess( args=list(args), returncode=1, stdout="", stderr="git is not installed" ) def _commit(self, message: str, allow_empty: bool = False) -> None: # Before the commit rather than after it: what a reader has to notice # is that the files changed, which is already true whether or not git # is there to record it. self.revision += 1 self._git("add", "-A") result = self._git( "-c", "user.name=fluksio", "-c", "user.email=fluksio@localhost", "commit", "-q", *(["--allow-empty"] if allow_empty else []), "-m", message, ) if result.returncode == 0 or "nothing to commit" in result.stdout: if self.revision % self._GC_EVERY == 0: # In the foreground, so this child is waited on rather than # orphaned. Plain `gc` rather than `gc --auto`, which the flag # above has just disabled. self._git("gc", "--quiet") return if result.stderr == "git is not installed": # Already said once, in `_git`. Repeating it per save is noise. return logger.warning("Could not commit flow change: %s", result.stdout.strip()) # ------------------------------------------------------------------------- # Paths # ------------------------------------------------------------------------- def _flow_dir(self, name: str) -> Path: return self.root / name def _flow_file(self, name: str) -> Path: return self._flow_dir(name) / "flow.json" def _draft_file(self, name: str) -> Path: return self._flow_dir(name) / "flow.draft.json" def _node_file(self, flow: str, node_id: str) -> Path: return self._flow_dir(flow) / "nodes" / f"{node_id}.py" def _draft_nodes_dir(self, flow: str) -> Path: return self._flow_dir(flow) / "nodes.draft" def _draft_node_file(self, flow: str, node_id: str) -> Path: return self._draft_nodes_dir(flow) / f"{node_id}.py" def _runtime_file(self, name: str) -> Path: return self._flow_dir(name) / "runtime.json" def _lib_file(self, name: str) -> Path: return self.root / LIB_DIR / f"{name}.py" def head(self) -> str: """The commit the store is at, so a result can name the code it ran.""" result = self._git("rev-parse", "HEAD") return result.stdout.strip() if result.returncode == 0 else "" # ------------------------------------------------------------------------- # Module requirements # # What node code may import, versioned with the flows importing it. A file # in the root cannot collide with a flow, which is always a directory. # ------------------------------------------------------------------------- def read_requirements(self) -> str: path = self.root / "requirements.txt" return path.read_text() if path.exists() else "" def write_requirements(self, text: str) -> None: (self.root / "requirements.txt").write_text(text) self._commit("Update module requirements") # ------------------------------------------------------------------------- # Shared node sources # # A node marked reusable moves its code to ``_lib/``; the flows using it # keep their own ports and parameters and point at it by name, so one fix # reaches all of them. ``_lib`` cannot collide with a flow — a flow name has # to start with a letter — and holds no ``flow.json``, so it is invisible to # everything that lists flows. # ------------------------------------------------------------------------- def list_lib(self) -> list[str]: return sorted(path.stem for path in (self.root / LIB_DIR).glob("*.py")) def lib_exists(self, name: str) -> bool: return self._lib_file(name).exists() def read_lib_source(self, name: str) -> str: path = self._lib_file(name) if not path.exists(): raise LibNotFound(name) return path.read_text() def write_lib_source(self, name: str, code: str) -> bool: path = self._lib_file(name) if path.exists() and path.read_text() == code: return False path.parent.mkdir(parents=True, exist_ok=True) path.write_text(code) self._commit(f"Update shared node '{name}'") return True def delete_lib_source(self, name: str) -> None: path = self._lib_file(name) if not path.exists(): raise LibNotFound(name) path.unlink() self._commit(f"Delete shared node '{name}'") def share_node(self, flow: str, node_id: str, lib_name: str) -> None: """Move a node's source into the library and point the node at it.""" with self._write_lock: if self.lib_exists(lib_name): raise LibExists(lib_name) code = self.read_node_source(flow, node_id, draft=True) self._lib_file(lib_name).parent.mkdir(parents=True, exist_ok=True) self._lib_file(lib_name).write_text(code) for path in ( self._node_file(flow, node_id), self._draft_node_file(flow, node_id), ): if path.exists(): path.unlink() self._set_source_ref(flow, node_id, lib_name) self._commit(f"Share node '{flow}.{node_id}' as '{lib_name}'") def unshare_node(self, flow: str, node_id: str) -> None: """Take a private copy of the shared source back into the flow.""" with self._write_lock: definition = self.read_flow(flow, draft=True) node = next((n for n in definition.nodes if n.id == node_id), None) if node is None or not node.source_ref: return code = self.read_lib_source(node.source_ref) # Written as a draft: the running flow keeps the shared code until # this is published, like every other edit. self._draft_node_file(flow, node_id).parent.mkdir( parents=True, exist_ok=True ) self._draft_node_file(flow, node_id).write_text(code) self._set_source_ref(flow, node_id, None) self._commit(f"Unshare node '{flow}.{node_id}'") def usages(self, lib_name: str) -> list[str]: """Every ``flow.node`` pointing at this shared source.""" used = [] for name in self.list_flows(): try: definition = self.read_flow(name, draft=True) except FlowNotFound: continue used += [ f"{name}.{node.id}" for node in definition.nodes if node.source_ref == lib_name ] return sorted(used) def _set_source_ref(self, flow: str, node_id: str, ref: str | None) -> None: """Point a node at a shared source, in the draft the editor is on.""" definition = self.read_flow(flow, draft=True) definition.nodes = [ node.model_copy(update={"source_ref": ref}) if node.id == node_id else node for node in definition.nodes ] stored = definition.model_copy(update={"version": definition.version + 1}) path = self._draft_file(flow) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(stored.model_dump_json(indent=2) + "\n") # ------------------------------------------------------------------------- # Runtime state # ------------------------------------------------------------------------- def read_enabled(self, name: str) -> bool: """Whether the engine should run this flow. Missing means yes.""" path = self._runtime_file(name) if not path.exists(): return True try: return bool(json.loads(path.read_text()).get("enabled", True)) except (ValueError, OSError): logger.warning("Unreadable runtime state for flow '%s'", name) return True def write_enabled(self, name: str, enabled: bool) -> None: """Stop or start a flow, in a file the editor never writes. This is deliberately not a field on the flow document: that one is autosaved from the canvas, so a stopped flow would start itself again on the next edit. """ path = self._runtime_file(name) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps({"enabled": enabled}, indent=2) + "\n") self._commit(f"{'Start' if enabled else 'Stop'} flow '{name}'") # ------------------------------------------------------------------------- # Flows # ------------------------------------------------------------------------- def list_flows(self) -> list[str]: """Every flow the editor knows, published or not.""" names = {path.parent.name for path in self.root.glob("*/flow.json")} names |= {path.parent.name for path in self.root.glob("*/flow.draft.json")} return sorted(names) def exists(self, name: str) -> bool: return self._flow_file(name).exists() or self._draft_file(name).exists() def is_published(self, name: str) -> bool: return self._flow_file(name).exists() def has_draft(self, name: str) -> bool: """Are there unpublished changes — to the structure or to any source?""" if self._draft_file(name).exists(): return True drafts = self._draft_nodes_dir(name) return drafts.exists() and any(drafts.glob("*.py")) def read_flow(self, name: str, draft: bool = False) -> FlowDef: """The published flow, or with ``draft`` the working copy.""" if draft: path = self._draft_file(name) if path.exists(): return FlowDef.model_validate_json(path.read_text()) path = self._flow_file(name) if not path.exists(): raise FlowNotFound(name) return FlowDef.model_validate_json(path.read_text()) def read_all(self) -> list[FlowDef]: """Every published flow — what the engine runs.""" flows = [] for path in sorted(self.root.glob("*/flow.json")): name = path.parent.name try: flows.append(self.read_flow(name)) except Exception: logger.exception("Skipping unreadable flow '%s'", name) return flows def write_flow(self, flow: FlowDef) -> bool: """Publish a flow directly. Returns False when nothing actually changed.""" path = self._flow_file(flow.name) content = flow.model_dump_json(indent=2) + "\n" if path.exists() and path.read_text() == content: return False path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content) self._commit(f"Update flow '{flow.name}'") return True def write_draft(self, flow: FlowDef, base_version: int) -> FlowDef: """Save unpublished changes, refusing to overwrite someone else's. ``base_version`` is the version the editor last saw. It has to match what is stored, otherwise another client saved in between and this write would silently discard their work. """ with self._write_lock: current: FlowDef | None = None if self.exists(flow.name): current = self.read_flow(flow.name, draft=True) if base_version != current.version: raise StaleVersion(flow.name, current.version) if _same_content(current, flow): return current stored = flow.model_copy( update={"version": (current.version if current else 0) + 1} ) path = self._draft_file(flow.name) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(stored.model_dump_json(indent=2) + "\n") self._commit(f"Update draft of flow '{flow.name}'") return stored def publish_flow(self, name: str, base_version: int) -> FlowDef: """Promote the working copy to what the engine runs.""" with self._write_lock: current = self.read_flow(name, draft=True) if base_version != current.version: raise StaleVersion(name, current.version) draft = self._draft_file(name) if draft.exists(): self._flow_file(name).write_text( current.model_dump_json(indent=2) + "\n" ) draft.unlink() drafts = self._draft_nodes_dir(name) if drafts.exists(): target = self._flow_dir(name) / "nodes" target.mkdir(parents=True, exist_ok=True) for source in sorted(drafts.glob("*.py")): shutil.copy2(source, target / source.name) shutil.rmtree(drafts) self._commit(f"Publish flow '{name}'") return current def discard_draft(self, name: str) -> FlowDef: """Throw the unpublished changes away and go back to what is running.""" with self._write_lock: draft = self._draft_file(name) if draft.exists(): draft.unlink() drafts = self._draft_nodes_dir(name) if drafts.exists(): shutil.rmtree(drafts) self._commit(f"Discard draft of flow '{name}'") return self.read_flow(name) def delete_flow(self, name: str) -> None: directory = self._flow_dir(name) if not directory.exists(): raise FlowNotFound(name) shutil.rmtree(directory) self._commit(f"Delete flow '{name}'") def rename_flow(self, name: str, new_name: str) -> FlowDef: """Rename a flow, carrying its nodes and any references to it. A flow's name is the namespace of its messages, so other flows reading ``old.temperature`` are rewritten to read ``new.temperature`` — leaving them pointing at a flow that no longer exists would break them silently. """ if not self.exists(name): raise FlowNotFound(name) if self.exists(new_name): raise FlowExists(new_name) self._flow_dir(name).rename(self._flow_dir(new_name)) # Published and draft alike carry the name, and both have to move. for path in (self._flow_file(new_name), self._draft_file(new_name)): if not path.exists(): continue flow = FlowDef.model_validate_json(path.read_text()) path.write_text( flow.model_copy(update={"name": new_name}).model_dump_json(indent=2) + "\n" ) for other in self.list_flows(): if other == new_name: continue for path in (self._flow_file(other), self._draft_file(other)): if not path.exists(): continue flow = FlowDef.model_validate_json(path.read_text()) if self._retarget(flow, f"{name}.", f"{new_name}."): path.write_text(flow.model_dump_json(indent=2) + "\n") self._commit(f"Rename flow '{name}' to '{new_name}'") return self.read_flow(new_name, draft=True) @staticmethod def _retarget(flow: FlowDef, old_prefix: str, new_prefix: str) -> bool: """Point this flow's cross-flow message names at a renamed flow.""" changed = False for node in flow.nodes: for specs in (node.requires, node.provides): for position, spec in enumerate(specs): if spec.name.startswith(old_prefix): tail = spec.name[len(old_prefix) :] specs[position] = spec.model_copy( update={"name": new_prefix + tail} ) changed = True return changed # ------------------------------------------------------------------------- # Node source # ------------------------------------------------------------------------- def read_node_source(self, flow: str, node_id: str, draft: bool = False) -> str: if draft: path = self._draft_node_file(flow, node_id) if path.exists(): return path.read_text() path = self._node_file(flow, node_id) return path.read_text() if path.exists() else DEFAULT_SOURCE def has_node_source(self, flow: str, node_id: str, draft: bool = False) -> bool: """Whether a body was ever stored for this node. `read_node_source` answers a template when there is none, which is what a new node's editor should open with — but it is a placeholder, not code anybody wrote, and telling the two apart is the difference between "someone edited this" and "this was never written". """ if draft and self._draft_node_file(flow, node_id).exists(): return True return self._node_file(flow, node_id).exists() def write_node_source( self, flow: str, node_id: str, code: str, draft: bool = False ) -> bool: if draft: # Compared against the working copy, so re-saving what is already # published does not create a draft file that never gets promoted. if self.read_node_source(flow, node_id, draft=True) == code: return False path = self._draft_node_file(flow, node_id) else: path = self._node_file(flow, node_id) if path.exists() and path.read_text() == code: return False path.parent.mkdir(parents=True, exist_ok=True) path.write_text(code) self._commit(f"Update {'draft of ' if draft else ''}node '{flow}.{node_id}'") return True def delete_node_source(self, flow: str, node_id: str) -> None: path = self._node_file(flow, node_id) if path.exists(): path.unlink() self._commit(f"Delete node '{flow}.{node_id}'")