feat: canonical flow directory with sync engine (files are source of truth)
This commit is contained in:
Jonas Miederer
parent
7d5ade6b6a
commit
c65e3ed332
16 changed files with 4167 additions and 42 deletions
274
backend/app/flows/sync.py
Normal file
274
backend/app/flows/sync.py
Normal file
|
|
@ -0,0 +1,274 @@
|
|||
"""Reconciliation: backend/flows/*.json (source of truth) -> DB + LFX.
|
||||
|
||||
The ONLY place that derives registry state from the canonical directory.
|
||||
Triggers (CLI, POST /api/flows/sync, embedded watcher) call
|
||||
sync_from_directory(); single-document callers (e.g. /import) use
|
||||
sync_document(). Everything is async (the LFX client is async) and guarded
|
||||
by a module lock so concurrent triggers serialize on SQLite's single writer.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
import uuid
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
from uuid import UUID
|
||||
|
||||
from sqlmodel import Session
|
||||
from watchfiles import DefaultFilter, awatch
|
||||
|
||||
from app.db import get_engine
|
||||
from app.models import FlowDefinition, FlowVersion
|
||||
from app.settings import settings
|
||||
|
||||
from . import export, store
|
||||
from .lfx_client import LfxClient, LfxError
|
||||
from .schemas import FlowDefinitionCreate, FlowVersionCreate, SyncReport
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Serializes all writers in this process (sync engine + manual API writes).
|
||||
sync_lock = threading.Lock()
|
||||
|
||||
|
||||
def lfx_id_for(definition_id: UUID, version_no: int) -> str:
|
||||
"""Deterministic LFX id (a real UUID: LFX validates the format).
|
||||
|
||||
Same (definition, version) always maps to the same id, so re-syncs and
|
||||
crash recovery overwrite instead of orphaning LFX registrations.
|
||||
"""
|
||||
return str(
|
||||
uuid.uuid5(uuid.NAMESPACE_URL, f"kirche-einfach/{definition_id}/v{version_no}")
|
||||
)
|
||||
|
||||
|
||||
def _latest_version(session: Session, flow_id: UUID) -> FlowVersion | None:
|
||||
versions = store.list_versions(session, flow_id)
|
||||
return versions[-1] if versions else None
|
||||
|
||||
|
||||
async def sync_document(
|
||||
session: Session,
|
||||
lfx: LfxClient,
|
||||
doc: dict,
|
||||
*,
|
||||
source_filename: str,
|
||||
author: str = "",
|
||||
deploy: bool = True,
|
||||
) -> tuple[FlowDefinition, FlowVersion | None, str]:
|
||||
"""Reconcile one parsed document. Returns (definition, version, action).
|
||||
|
||||
Actions: "added" (new definition + version), "updated" (new version),
|
||||
"skipped" (content unchanged). Raises store.ConflictError on name
|
||||
clashes, store.InvalidError on invalid JSON, LfxError when deploy fails.
|
||||
With deploy=False a version is recorded without touching LFX.
|
||||
"""
|
||||
store.validate_flow_json(doc)
|
||||
name = doc.get("name") or Path(source_filename).stem
|
||||
with sync_lock:
|
||||
definition = store.get_definition_by_source_path(session, source_filename)
|
||||
if definition is None:
|
||||
definition = store.create_definition(
|
||||
session, FlowDefinitionCreate(name=name)
|
||||
)
|
||||
definition.source_path = source_filename
|
||||
session.add(definition)
|
||||
session.commit()
|
||||
session.refresh(definition)
|
||||
action = "added"
|
||||
else:
|
||||
action = "updated"
|
||||
if definition.name != name:
|
||||
store.set_definition_name(session, definition, name)
|
||||
|
||||
latest = _latest_version(session, definition.id)
|
||||
if (
|
||||
latest is not None
|
||||
and export.content_hash(latest.langflow_json)
|
||||
== export.content_hash(doc)
|
||||
):
|
||||
return definition, latest, "skipped"
|
||||
|
||||
if not deploy:
|
||||
version = store.create_version(
|
||||
session,
|
||||
definition.id,
|
||||
FlowVersionCreate(langflow_json=doc, author=author),
|
||||
)
|
||||
return definition, version, action
|
||||
|
||||
version_no = store.next_version_no(session, definition.id)
|
||||
lfx_id = lfx_id_for(definition.id, version_no)
|
||||
uploaded = await lfx.upload_flow(doc, flow_id=lfx_id, replace=True)
|
||||
version = store.create_version(
|
||||
session,
|
||||
definition.id,
|
||||
FlowVersionCreate(langflow_json=doc, author=author),
|
||||
)
|
||||
if version.version_no != version_no: # pragma: no cover - lock prevents this
|
||||
raise store.ConflictError(
|
||||
f"concurrent version write for flow {definition.id}"
|
||||
)
|
||||
version = store.mark_deployed(session, version, uploaded.id)
|
||||
if (
|
||||
latest is not None
|
||||
and latest.lfx_flow_id
|
||||
and latest.lfx_flow_id != uploaded.id
|
||||
):
|
||||
try:
|
||||
await lfx.delete_flow(latest.lfx_flow_id)
|
||||
except LfxError as exc:
|
||||
logger.warning(
|
||||
"LFX unregister %s failed (continuing): %s",
|
||||
latest.lfx_flow_id,
|
||||
exc,
|
||||
)
|
||||
return definition, version, action
|
||||
|
||||
|
||||
async def delete_definition_tree(
|
||||
session: Session, lfx: LfxClient, definition: FlowDefinition
|
||||
) -> dict[str, int]:
|
||||
"""Remove a definition whose canonical file is gone.
|
||||
|
||||
LFX registrations go first (best-effort, fail-closed for new runs), the
|
||||
DB subtree (tasks, runs, versions, definition) after. Deterministic LFX
|
||||
ids make any crash in between self-healing on the next sync.
|
||||
"""
|
||||
for version in store.list_versions(session, definition.id):
|
||||
if version.lfx_flow_id:
|
||||
try:
|
||||
await lfx.delete_flow(version.lfx_flow_id)
|
||||
except LfxError as exc:
|
||||
logger.warning(
|
||||
"LFX unregister %s failed (continuing): %s",
|
||||
version.lfx_flow_id,
|
||||
exc,
|
||||
)
|
||||
with sync_lock:
|
||||
return store.delete_definition_cascade(session, definition)
|
||||
|
||||
|
||||
async def sync_from_directory(
|
||||
session: Session, lfx: LfxClient, flows_dir: str | Path
|
||||
) -> SyncReport:
|
||||
"""Reconcile the whole canonical directory. Per-file problems land in
|
||||
report.errors (never raised); a missing directory raises FileNotFoundError
|
||||
so misconfiguration fails loudly instead of wiping the registry.
|
||||
"""
|
||||
flows_dir = Path(flows_dir)
|
||||
report = SyncReport()
|
||||
if not flows_dir.is_dir():
|
||||
raise FileNotFoundError(f"flows directory missing: {flows_dir}")
|
||||
files = {path.name: path for path in export.iter_flow_files(flows_dir)}
|
||||
for filename, path in files.items():
|
||||
try:
|
||||
doc = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError) as exc:
|
||||
report.errors[filename] = f"unreadable: {exc}"
|
||||
continue
|
||||
if not isinstance(doc, dict):
|
||||
report.errors[filename] = "top-level JSON must be an object"
|
||||
continue
|
||||
try:
|
||||
_, _, action = await sync_document(
|
||||
session, lfx, doc, source_filename=filename
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001 - reconciliation is total
|
||||
report.errors[filename] = str(exc)
|
||||
continue
|
||||
getattr(report, action).append(path.stem)
|
||||
known = {
|
||||
definition.source_path
|
||||
for definition in store.list_definitions(session)
|
||||
if definition.source_path
|
||||
}
|
||||
for stale in sorted(known - set(files)):
|
||||
definition = store.get_definition_by_source_path(session, stale)
|
||||
if definition is None: # pragma: no cover - defensive
|
||||
continue
|
||||
try:
|
||||
await delete_definition_tree(session, lfx, definition)
|
||||
except Exception as exc: # noqa: BLE001 - reconciliation is total
|
||||
report.errors[stale] = str(exc)
|
||||
continue
|
||||
report.deleted.append(Path(stale).stem)
|
||||
return report
|
||||
|
||||
|
||||
def default_session_factory() -> Callable[[], Session]:
|
||||
return lambda: Session(get_engine())
|
||||
|
||||
|
||||
def default_lfx_factory() -> Callable[[], LfxClient]:
|
||||
return lambda: LfxClient(
|
||||
settings.lfx_url, api_key=settings.lfx_api_key or None
|
||||
)
|
||||
|
||||
|
||||
def run_sync_blocking(
|
||||
flows_dir: str | Path,
|
||||
*,
|
||||
session_factory: Callable[[], Session] | None = None,
|
||||
lfx_factory: Callable[[], LfxClient] | None = None,
|
||||
) -> SyncReport:
|
||||
"""Blocking one-shot sync for worker threads (lifespan, watcher)."""
|
||||
session = (session_factory or default_session_factory())()
|
||||
try:
|
||||
lfx = (lfx_factory or default_lfx_factory())()
|
||||
return asyncio.run(sync_from_directory(session, lfx, Path(flows_dir)))
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
class _JsonFilter(DefaultFilter):
|
||||
def __call__(self, change, path: str) -> bool: # type: ignore[override]
|
||||
return super().__call__(change, path) and path.endswith(".json")
|
||||
|
||||
|
||||
async def watch_forever(
|
||||
stop: asyncio.Event,
|
||||
flows_dir: str | Path,
|
||||
run_sync: Callable[[], SyncReport] | None = None,
|
||||
) -> None:
|
||||
"""Run until `stop` is set: sync on every batch of .json changes.
|
||||
|
||||
All failures are logged, never raised — a broken file must not kill the
|
||||
watcher. `run_sync` is injectable for tests.
|
||||
"""
|
||||
flows_dir = Path(flows_dir)
|
||||
run = run_sync or (lambda: run_sync_blocking(flows_dir))
|
||||
while not stop.is_set() and not flows_dir.is_dir():
|
||||
logger.error("flows dir missing, retrying in 30s: %s", flows_dir)
|
||||
try:
|
||||
await asyncio.wait_for(stop.wait(), timeout=30)
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
if stop.is_set():
|
||||
return
|
||||
logger.info("watching %s", flows_dir)
|
||||
async for _ in awatch(
|
||||
flows_dir, watch_filter=_JsonFilter(), debounce=1000, step=100
|
||||
):
|
||||
if stop.is_set():
|
||||
break
|
||||
try:
|
||||
report = await asyncio.to_thread(run)
|
||||
except FileNotFoundError as exc:
|
||||
logger.error("flow sync failed: %s", exc)
|
||||
continue
|
||||
except Exception: # noqa: BLE001 - watcher must survive
|
||||
logger.exception("flow sync failed")
|
||||
continue
|
||||
logger.info(
|
||||
"flow sync: +%d ~%d -%d skipped=%d errors=%d",
|
||||
len(report.added),
|
||||
len(report.updated),
|
||||
len(report.deleted),
|
||||
len(report.skipped),
|
||||
len(report.errors),
|
||||
)
|
||||
Loading…
Add table
Add a link
Reference in a new issue