| 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") | |