65 lines
2 KiB
Python
65 lines
2 KiB
Python
"""Canonical flow files: backend/flows/<name>.json is the source of truth.
|
|
|
|
One Langflow export per file, committed to git. This module only reads and
|
|
writes the directory; reconciliation into the DB lives in sync.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
FLOWS_GLOB = "*.json"
|
|
|
|
|
|
def slugify(name: str) -> str:
|
|
"""Map a flow name to a safe filename stem (lowercase, [a-z0-9-])."""
|
|
slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")
|
|
return slug or "flow"
|
|
|
|
|
|
def canonical_bytes(doc: dict) -> bytes:
|
|
"""Deterministic serialization: stable bytes for stable git diffs."""
|
|
return (json.dumps(doc, indent=2, ensure_ascii=False, sort_keys=True) + "\n").encode("utf-8")
|
|
|
|
|
|
def content_hash(doc: dict) -> str:
|
|
"""Hash of the parsed document (formatting-independent)."""
|
|
return hashlib.sha256(canonical_bytes(doc)).hexdigest()
|
|
|
|
|
|
def flow_path(flows_dir: str | Path, stem: str) -> Path:
|
|
return Path(flows_dir) / f"{stem}.json"
|
|
|
|
|
|
def write_flow_file(flows_dir: str | Path, doc: dict, *, name: str | None = None) -> Path:
|
|
"""Write a flow document to backend/flows atomically (tmp + rename).
|
|
|
|
The filename derives from `name` (or the document's own name). Returns
|
|
the written path.
|
|
"""
|
|
flows_dir = Path(flows_dir)
|
|
flows_dir.mkdir(parents=True, exist_ok=True)
|
|
stem = slugify(name or doc.get("name") or "flow")
|
|
target = flow_path(flows_dir, stem)
|
|
fd, tmp_name = tempfile.mkstemp(dir=str(flows_dir), prefix=f".{stem}-", suffix=".tmp")
|
|
try:
|
|
with os.fdopen(fd, "wb") as tmp:
|
|
tmp.write(canonical_bytes(doc))
|
|
os.replace(tmp_name, target)
|
|
except BaseException:
|
|
try:
|
|
os.unlink(tmp_name)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
return target
|
|
|
|
|
|
def iter_flow_files(flows_dir: str | Path) -> list[Path]:
|
|
"""Sorted list of candidate flow files (top level only)."""
|
|
return sorted(Path(flows_dir).glob(FLOWS_GLOB))
|