Create persistence.py

engine_03/flow/persistence.py

@@ -0,0 +1,25 @@
+import json
+from pathlib import Path
+from typing import Union
+
+from .graph_models import FlowGraph, Node, Edge
+
+
+def load_flow(path: Union[str, Path]) -> FlowGraph:
+    path = Path(path)
+    if not path.exists():
+        raise FileNotFoundError(path)
+
+    data = json.load(path.open("r", encoding="utf-8"))
+    nodes = [Node(**n) for n in data.get("nodes", [])]
+    edges = [Edge(**e) for e in data.get("edges", [])]
+    return FlowGraph(nodes=nodes, edges=edges)
+
+
+def save_flow(flow: FlowGraph, path: Union[str, Path]) -> None:
+    path = Path(path)
+    data = {
+        "nodes": [n.__dict__ for n in flow.nodes],
+        "edges": [e.__dict__ for e in flow.edges],
+    }
+    path.write_text(json.dumps(data, indent=2), encoding="utf-8")