782 lines
32 KiB
Python
782 lines
32 KiB
Python
"""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 collections.abc import Iterable
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from fluksio.flow.schemas import FlowDef, _validate_name
|
|
|
|
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 InvalidName(ValueError):
|
|
"""A name that cannot be a file or directory of the store."""
|
|
|
|
def __init__(self, name: str) -> None:
|
|
super().__init__(name)
|
|
self.name = name
|
|
|
|
def __str__(self) -> str:
|
|
return (
|
|
f"'{self.name}' is not a valid name: use lowercase letters, digits "
|
|
"and underscores, starting with a letter"
|
|
)
|
|
|
|
|
|
def checked(name: str) -> str:
|
|
"""*name*, once it is certain to be one path segment inside the store.
|
|
|
|
Every path the stores build starts from a name the API was handed,
|
|
percent-decoded on the way in, so without this `..` or `_lib` reached the
|
|
filesystem as written: `delete_flow("..")` removed the data directory.
|
|
"""
|
|
try:
|
|
return _validate_name(name)
|
|
except ValueError:
|
|
raise InvalidName(name) from None
|
|
|
|
|
|
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 _updated_at(paths: Iterable[Path]) -> datetime | None:
|
|
"""When the newest of these files was written, or ``None`` if none exist.
|
|
|
|
A file's mtime, not the commit that recorded it: every write here commits
|
|
immediately, so the two are the same instant, and `git log -1 -- <path>`
|
|
would walk the history back to the last commit touching that path — 130ms
|
|
for a document nobody has edited in a week, against a `stat` for all of
|
|
them. Slowest for exactly the stalest documents is the wrong shape for a
|
|
list endpoint.
|
|
"""
|
|
stamps = [path.stat().st_mtime for path in paths if path.exists()]
|
|
return datetime.fromtimestamp(max(stamps), UTC) if stamps else None
|
|
|
|
|
|
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
|
|
#: `read_all` memoised against that revision — see the method.
|
|
self._read_all: list[FlowDef] = []
|
|
self._read_all_at = -1
|
|
self._read_all_lock = threading.Lock()
|
|
if not (self.root / ".git").exists():
|
|
self._git("init", "-q")
|
|
self._commit("Initialise flow store", allow_empty=True)
|
|
self._migrate()
|
|
|
|
def _migrate(self) -> None:
|
|
"""Rewrite node types and settings stored under a name they no longer have.
|
|
|
|
On 2026-10-15 the node type ``trigger`` became ``debounce`` and the mqtt
|
|
setting ``publish_queue_size`` became ``send_queue_size``, with no alias
|
|
left for the old names. Each published and draft document is read as
|
|
plain JSON and only the matching node changes, so a field this version
|
|
does not model survives, and the old document stays in git. The version
|
|
goes up by one, so an editor still holding the old document gets a
|
|
conflict on its next save instead of writing the old name back. A store
|
|
with nothing to rename is left alone.
|
|
"""
|
|
changed = []
|
|
paths = [*self.root.glob("*/flow.json"), *self.root.glob("*/flow.draft.json")]
|
|
for path in sorted(paths):
|
|
try:
|
|
data = json.loads(path.read_text())
|
|
nodes = [n for n in data["nodes"] if n.get("type") == "trigger"]
|
|
queues = [
|
|
n["params"]
|
|
for n in data["nodes"]
|
|
if n.get("type") == "mqtt"
|
|
and "publish_queue_size" in (n.get("params") or {})
|
|
]
|
|
except Exception:
|
|
# Unreadable to the engine too, which reports it when it loads.
|
|
continue
|
|
if not nodes and not queues:
|
|
continue
|
|
for node in nodes:
|
|
node["type"] = "debounce"
|
|
for params in queues:
|
|
params["send_queue_size"] = params.pop("publish_queue_size")
|
|
data["version"] = data.get("version", 1) + 1
|
|
path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n")
|
|
changed.append(f"{path.parent.name}/{path.name}")
|
|
if changed:
|
|
logger.warning(
|
|
"Renamed node type 'trigger' to 'debounce' and mqtt "
|
|
"'publish_queue_size' to 'send_queue_size' in %s",
|
|
", ".join(changed),
|
|
)
|
|
self._commit("Rename 'trigger' nodes and mqtt 'publish_queue_size'")
|
|
|
|
# -------------------------------------------------------------------------
|
|
# 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
|
|
# -------------------------------------------------------------------------
|
|
|
|
# Every path below starts in one of these three, and each name in it has
|
|
# been through `checked`.
|
|
|
|
def _flow_dir(self, name: str) -> Path:
|
|
return self.root / checked(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"{checked(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"{checked(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"{checked(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 detach_lib(self, name: str) -> None:
|
|
"""Give every node running this shared source a copy, then delete it.
|
|
|
|
Published flows and drafts alike, so a running flow keeps the same code
|
|
and nothing waits on a publish. Each document changed goes up a version,
|
|
so an editor holding the old one gets a conflict on its next save
|
|
instead of pointing a node back at a source that is gone.
|
|
"""
|
|
with self._write_lock:
|
|
code = self.read_lib_source(name)
|
|
for flow in self.list_flows():
|
|
for path, node_file in (
|
|
(self._flow_file(flow), self._node_file),
|
|
(self._draft_file(flow), self._draft_node_file),
|
|
):
|
|
if not path.exists():
|
|
continue
|
|
definition = FlowDef.model_validate_json(path.read_text())
|
|
users = [n for n in definition.nodes if n.source_ref == name]
|
|
if not users:
|
|
continue
|
|
for node in users:
|
|
target = node_file(flow, node.id)
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text(code)
|
|
node.source_ref = None
|
|
definition.version += 1
|
|
path.write_text(definition.model_dump_json(indent=2) + "\n")
|
|
self._lib_file(name).unlink()
|
|
self._commit(f"Detach and delete shared node '{name}'")
|
|
|
|
def usages(self, lib_name: str, draft: bool = True) -> list[str]:
|
|
"""Every ``flow.node`` pointing at this shared source.
|
|
|
|
Read from the drafts the editor shows, or with ``draft`` off from the
|
|
published flows the engine runs.
|
|
"""
|
|
used = []
|
|
for name in self.list_flows():
|
|
try:
|
|
definition = self.read_flow(name, draft=draft)
|
|
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 updated_at(self, name: str) -> datetime | None:
|
|
"""When this flow was last written — its document or any node's source.
|
|
|
|
Node code is saved without touching the flow document, so a flow whose
|
|
last change was to a node body would otherwise look untouched.
|
|
"""
|
|
return _updated_at(
|
|
[self._flow_file(name), self._draft_file(name)]
|
|
+ list((self._flow_dir(name) / "nodes").glob("*.py"))
|
|
+ list(self._draft_nodes_dir(name).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.
|
|
|
|
Cached against `revision`, because this sits on the publish path: a
|
|
dashboard slider moving asked every flow's file to be read and
|
|
validated again, per value. The list is rebuilt on the next commit,
|
|
and returned as a copy so a caller sorting it cannot disturb the next.
|
|
"""
|
|
with self._read_all_lock:
|
|
if self._read_all_at != self.revision:
|
|
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)
|
|
self._read_all = flows
|
|
self._read_all_at = self.revision
|
|
return list(self._read_all)
|
|
|
|
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)
|
|
|
|
def duplicate_flow(self, name: str) -> FlowDef:
|
|
"""Copy a flow's working copy into a new flow that is only a draft.
|
|
|
|
Like a new flow, the copy runs nothing until someone publishes it. It is
|
|
named ``{name}_copy``, then ``_copy_2`` and on. ``origin`` is dropped,
|
|
since the copy is a local flow ``fluksio sync`` must not claim, and
|
|
``runtime.json`` stays behind with everything else about running.
|
|
"""
|
|
with self._write_lock:
|
|
source = self.read_flow(name, draft=True)
|
|
new_name, count = f"{name}_copy", 1
|
|
# The directory rather than `exists`: anything left in it would
|
|
# otherwise end up in the copy.
|
|
while self._flow_dir(new_name).exists():
|
|
count += 1
|
|
new_name = f"{name}_copy_{count}"
|
|
|
|
copy = source.model_copy(
|
|
update={
|
|
"name": new_name,
|
|
"title": f"{source.title} (copy)" if source.title else "",
|
|
"version": 1,
|
|
"origin": None,
|
|
}
|
|
)
|
|
# Its own messages named with its own name are the copy's now.
|
|
self._retarget(copy, f"{name}.", f"{new_name}.")
|
|
path = self._draft_file(new_name)
|
|
path.parent.mkdir(parents=True)
|
|
path.write_text(copy.model_dump_json(indent=2) + "\n")
|
|
|
|
# The bodies the source's editor shows: published, edits over them.
|
|
# As drafts, so publishing the copy promotes them like a new flow's.
|
|
target = self._draft_nodes_dir(new_name)
|
|
for folder in (self._flow_dir(name) / "nodes", self._draft_nodes_dir(name)):
|
|
for body in sorted(folder.glob("*.py")):
|
|
target.mkdir(exist_ok=True)
|
|
shutil.copy2(body, target / body.name)
|
|
|
|
self._commit(f"Duplicate flow '{name}' as '{new_name}'")
|
|
return copy
|
|
|
|
@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}'")
|