88 lines
2.9 KiB
Python
88 lines
2.9 KiB
Python
"""Reconcile backend/flows/*.json into the DB + LFX.
|
|
|
|
Usage (from backend/, venv active):
|
|
uv run python scripts/sync_flows.py --sync [--dir backend/flows]
|
|
uv run python scripts/sync_flows.py --pull [--sync] # export all Langflow flows to files first
|
|
|
|
Exit codes: 0 ok, 1 sync errors, 2 flows dir missing.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from sqlmodel import Session
|
|
|
|
from app.db import get_engine, reset_engine
|
|
from app.flows import export, sync
|
|
from app.flows.langflow_client import LangflowClient, LangflowError
|
|
from app.flows.lfx_client import LfxClient
|
|
from app.flows.schemas import SyncReport
|
|
from app.settings import settings
|
|
|
|
|
|
async def pull_from_langflow(flows_dir: Path) -> dict:
|
|
"""Export every Langflow flow to its own canonical file. Returns a report."""
|
|
client = LangflowClient(settings.langflow_url, api_key=settings.langflow_api_key)
|
|
headers = await client.list_flows()
|
|
pulled: list[str] = []
|
|
errors: dict[str, str] = {}
|
|
for header in headers:
|
|
flow_id = str(header.get("id", ""))
|
|
try:
|
|
doc = await client.export_flow(flow_id)
|
|
name = doc.get("name") or flow_id
|
|
path = export.write_flow_file(flows_dir, doc, name=name)
|
|
pulled.append(path.stem)
|
|
except (LangflowError, OSError, ValueError) as exc:
|
|
errors[flow_id] = str(exc)
|
|
return {"pulled": sorted(pulled), "errors": errors}
|
|
|
|
|
|
async def run_sync(flows_dir: Path) -> SyncReport:
|
|
reset_engine()
|
|
session = Session(get_engine())
|
|
try:
|
|
lfx = LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None)
|
|
return await sync.sync_from_directory(session, lfx, flows_dir)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
async def main_async(args) -> int:
|
|
flows_dir = Path(args.dir or settings.flows_dir)
|
|
if args.pull:
|
|
pulled = await pull_from_langflow(flows_dir)
|
|
print(json.dumps(pulled, indent=2, ensure_ascii=False))
|
|
if pulled["errors"]:
|
|
return 1
|
|
if args.sync or not args.pull:
|
|
try:
|
|
report = await run_sync(flows_dir)
|
|
except FileNotFoundError as exc:
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
return 2
|
|
print(report.model_dump_json(indent=2))
|
|
return 1 if report.errors else 0
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--pull", action="store_true", help="export all Langflow flows to backend/flows first"
|
|
)
|
|
parser.add_argument("--sync", action="store_true", help="reconcile backend/flows into DB + LFX")
|
|
parser.add_argument("--dir", default=None, help="flows directory (default: settings.flows_dir)")
|
|
args = parser.parse_args()
|
|
if not args.pull and not args.sync:
|
|
args.sync = True
|
|
return asyncio.run(main_async(args))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|