Compare commits
4 commits
acea6ae624
...
0b03399c1a
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0b03399c1a | ||
|
|
3ee997cb72 | ||
|
|
95a1c1afb7 | ||
|
|
1326cece42 |
20 changed files with 5081 additions and 3911 deletions
24
.env.example
24
.env.example
|
|
@ -17,6 +17,11 @@ LANGFLOW_SUPERUSER=langflow
|
|||
# Long random values (check_prod.py enforces length). Never leave the
|
||||
# placeholders: the authoring UI is internet-facing.
|
||||
LANGFLOW_SUPERUSER_PASSWORD=
|
||||
# MUST be a valid Fernet key (44 chars, urlsafe-base64, 32 bytes) — any
|
||||
# other value breaks variable decryption ("Fernet key must be 32
|
||||
# url-safe base64-encoded bytes") and provider keys cannot be saved.
|
||||
# Generate with:
|
||||
# python -c "import base64,os;print(base64.urlsafe_b64encode(os.urandom(32)).decode())"
|
||||
LANGFLOW_SECRET_KEY=
|
||||
# Single static key; validated against env (LANGFLOW_API_KEY_SOURCE=env), superuser scope.
|
||||
LANGFLOW_API_KEY=sk-
|
||||
|
|
@ -51,4 +56,21 @@ SMTP_USERNAME=
|
|||
SMTP_PASSWORD=
|
||||
|
||||
# --- LLM (optional; tests mock the provider without it) ---
|
||||
# LLM_API_KEY=
|
||||
# LLM_API_KEY=
|
||||
|
||||
# --- LLM provider für Langflow (Authoring-UI + LFX-Runtime) ---
|
||||
# Langflow erzeugt daraus automatisch Global Variables (Credential-Typ);
|
||||
# LFX liest dieselben Werte zur Laufzeit. Eine in der UI angelegte Global
|
||||
# Variable mit gleichem Namen hat Vorrang, die Env-Variable ist Fallback.
|
||||
# OpenAI-kompatibler Endpoint (auch selbst gehostet); ohne eigenen Provider
|
||||
# leer lassen. Beispiel: https://openrouter.ai/api/v1
|
||||
OPENAI_API_KEY=
|
||||
OPENAI_BASE_URL=
|
||||
|
||||
# --- Langflow-Login-Sitzung ---
|
||||
# Refresh-Token-Gültigkeit in Sekunden (Default 604800 = 7 Tage).
|
||||
# 7776000 = 90 Tage: einmal einloggen, Container-Restarts übersteht die
|
||||
# Session, solange LANGFLOW_SECRET_KEY stabil bleibt. In Prod kürzer wählen.
|
||||
# LANGFLOW_REFRESH_TOKEN_EXPIRE_SECONDS=7776000
|
||||
# Erstellt Global Variables aus Env (Credential-Typ), restart-fest:
|
||||
# LANGFLOW_VARIABLES_TO_GET_FROM_ENVIRONMENT=OPENAI_API_KEY,OPENAI_BASE_URL
|
||||
30
CHANGELOG.md
30
CHANGELOG.md
|
|
@ -5,7 +5,37 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
|
|||
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||
|
||||
## [Unreleased]
|
||||
### Added
|
||||
- Flow-Sync ist jetzt ein manueller Full-Pass: `POST /api/flows/sync`
|
||||
(Button "Aus Langflow synchronisieren" in der Verwaltung) holt erst alle
|
||||
Flows aus Langflow nach `backend/flows/*.json` und deployed sie danach
|
||||
automatisch (Report mit `pulled`-Liste). Der Hintergrund-Beobachter und der
|
||||
Startup-Sync bleiben dateibasiert. Die Verwaltung zeigt eine deutsche
|
||||
Schritt-Anleitung direkt beim Sync-Button.
|
||||
- LLM-Provider-Umgebung für Langflow: `OPENAI_API_KEY` + `OPENAI_BASE_URL`
|
||||
in `.env` (OpenAI-kompatible Endpoints) erzeugen automatisch
|
||||
restart-feste Global Variables in Langflow; LFX liest dieselben Werte.
|
||||
- Langflow-Sitzungen halten länger: `LANGFLOW_REFRESH_TOKEN_EXPIRE_SECONDS`
|
||||
(Default weiter 7 Tage, Dev-Beispiel 90) + stabile JWT-Signatur via
|
||||
`LANGFLOW_SECRET_KEY` — kein Re-Login nach Container-Restarts.
|
||||
### Fixed
|
||||
- Flow-Sync aus Langflow: neue Flows blieben unsichtbar, weil LFX-Uploads
|
||||
mit `Permission denied` scheiterten (named Volume root-owned, LFX läuft
|
||||
rootless). `lfx-entrypoint.sh` chownt `/flows` per gosu vor dem Start;
|
||||
zusätzlich Pull-Dedup (wiederverwendet gleichnamige/identische Dateien
|
||||
statt Bindestrich-Duplikate wie `simple-rule.json`, die jede Sync-Meldung
|
||||
mit Namenskonflikten füllten).
|
||||
- Sync-Dateinamen behalten die Langflow-Namen (`test_flow.json` statt
|
||||
`test-flow.json`; nur dateisystem-reservierte Zeichen werden ersetzt).
|
||||
- Gezogene Flow-Dateien sind world-readable (0o644): `mkstemp` erzeugte
|
||||
0o600-Dateien, die Langflow beim Start abstürzen ließen
|
||||
(`Permission denied` in `/app/flows`, Crash-Loop) und der API Fehler
|
||||
(`unreadable`) verursachten.
|
||||
- `LANGFLOW_SECRET_KEY` muss ein gültiger Fernet-Key sein (44 Zeichen,
|
||||
urlsafe-base64): ein Hex-String brach die Variablen-Verschlüsselung mit
|
||||
`Fernet key must be 32 url-safe base64-encoded bytes` (500 beim Anlegen
|
||||
von Global Variables — Provider-Keys ließen sich nicht speichern).
|
||||
`.env.example` dokumentiert die Erzeugung.
|
||||
- LFX runtime user gets a home directory (`useradd -m`, `HOME` set):
|
||||
without one every flow run failed with "Settings service not found".
|
||||
### Security
|
||||
|
|
|
|||
|
|
@ -16,11 +16,18 @@ RUN pip install "lfx~=${LFX_VERSION}" requests
|
|||
# Non-root runtime: uploads persist into the mounted volume. The user needs
|
||||
# a home directory: lfx resolves settings/cache paths from $HOME, and
|
||||
# without one every flow run fails ("Settings service not found").
|
||||
# gosu drops privileges in the entrypoint after chowning /flows: the named
|
||||
# volume arrives root-owned and the server user could otherwise never write
|
||||
# uploads (every deploy would fail with "Permission denied").
|
||||
ENV HOME=/home/lfx
|
||||
RUN useradd -r -m -s /bin/false lfx && mkdir -p /flows && chown lfx /flows
|
||||
RUN apt-get update \
|
||||
&& apt-get install -y --no-install-recommends gosu \
|
||||
&& rm -rf /var/lib/apt/lists/* \
|
||||
&& useradd -r -m -s /bin/false lfx && mkdir -p /flows && chown lfx /flows
|
||||
COPY lfx-entrypoint.sh /usr/local/bin/lfx-entrypoint.sh
|
||||
VOLUME /flows
|
||||
EXPOSE 6004
|
||||
USER lfx
|
||||
ENTRYPOINT ["sh", "/usr/local/bin/lfx-entrypoint.sh"]
|
||||
|
||||
# Serve the /flows dir (uploaded flows persist there via --flow-dir).
|
||||
# No positional path: with one, lfx>=1.12.1 exits when the dir holds no
|
||||
|
|
|
|||
15
README.md
15
README.md
|
|
@ -89,7 +89,11 @@ a database server. Postgres can return later by changing `DATABASE_URL`
|
|||
|
||||
`.env` holds only ubiquitous variables (identical value in dev and prod:
|
||||
secrets, service URLs). Dev-only variables travel via `devcontainer.json`
|
||||
(`APP_ENV=dev`); paths resolve from the repo root in code.
|
||||
(`APP_ENV=dev`); paths resolve from the repo root in code. LLM provider
|
||||
keys for Langflow are plain env vars (`OPENAI_API_KEY`,
|
||||
`OPENAI_BASE_URL`) — Langflow turns them into restart-proof global
|
||||
variables, LFX reads them at runtime. `LANGFLOW_SECRET_KEY` must be a
|
||||
valid Fernet key (see `.env.example`).
|
||||
|
||||
## Getting Started
|
||||
|
||||
|
|
@ -122,9 +126,12 @@ Frontend (node lives in the `frontend` container):
|
|||
- `docker compose exec frontend npm run build` — typecheck + build.
|
||||
- Without a docker socket: VS Code "Attach to Running Container" → frontend.
|
||||
|
||||
Flow workflow: design in Langflow → export into `backend/flows/<name>.json`
|
||||
(`--pull`, or UI download) → commit → the watcher deploys it automatically.
|
||||
Deleting a file removes the flow including its runs (git keeps the JSON).
|
||||
Flow workflow: design in Langflow → press "Aus Langflow synchronisieren"
|
||||
in Verwaltung (manual full pass: Langflow → `backend/flows/<name>.json` →
|
||||
DB + LFX deploy) → commit. File-only alternatives: `scripts/sync_flows.py
|
||||
--pull --sync`, UI download, or the single-flow import by Langflow ID. The
|
||||
watcher deploys file edits automatically. Deleting a file removes the flow
|
||||
including its runs (git keeps the JSON).
|
||||
|
||||
## Production (self-hosted — TLS/reverse proxy required, not included)
|
||||
|
||||
|
|
|
|||
|
|
@ -29,7 +29,7 @@ from app.settings import settings
|
|||
|
||||
from . import input_schema as form_schema
|
||||
from . import runner, store
|
||||
from .deploy import import_from_langflow
|
||||
from .deploy import import_from_langflow, pull_all_from_langflow
|
||||
from .langflow_client import (
|
||||
LangflowClient,
|
||||
LangflowError,
|
||||
|
|
@ -160,13 +160,25 @@ def list_flows(session: Annotated[Session, Depends(get_db)]):
|
|||
async def sync_flows(
|
||||
session: Annotated[Session, Depends(get_db)],
|
||||
lfx: Annotated[LfxClient, Depends(get_lfx)],
|
||||
langflow: Annotated[LangflowClient, Depends(get_langflow)],
|
||||
_admin: Annotated[User, Depends(require_admin)],
|
||||
):
|
||||
"""Reconcile backend/flows/*.json into the DB + LFX (add/update/delete)."""
|
||||
"""Manual full pass: Langflow -> backend/flows/*.json -> DB + LFX."""
|
||||
try:
|
||||
return await sync_from_directory(session, lfx, Path(settings.flows_dir))
|
||||
pulled_result = await pull_all_from_langflow(langflow, Path(settings.flows_dir))
|
||||
except LangflowUnavailable as exc:
|
||||
logger.warning("langflow unavailable: %s", exc)
|
||||
raise HTTPException(status_code=503, detail="authoring server unavailable") from exc
|
||||
except LangflowError as exc:
|
||||
logger.warning("langflow error: %s", exc)
|
||||
raise HTTPException(status_code=502, detail="authoring server error") from exc
|
||||
try:
|
||||
report = await sync_from_directory(session, lfx, Path(settings.flows_dir))
|
||||
except FileNotFoundError as exc:
|
||||
raise HTTPException(status_code=500, detail=str(exc)) from exc
|
||||
report.pulled = pulled_result["pulled"]
|
||||
report.errors = {**pulled_result["errors"], **report.errors}
|
||||
return report
|
||||
|
||||
|
||||
@flows_router.get("/{flow_id}", response_model=FlowDefinitionRead)
|
||||
|
|
|
|||
|
|
@ -2,16 +2,65 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from sqlmodel import Session
|
||||
|
||||
from app.models import FlowDefinition, FlowVersion
|
||||
from app.settings import settings
|
||||
|
||||
from . import export, sync
|
||||
from .langflow_client import LangflowClient
|
||||
from .langflow_client import LangflowClient, LangflowError
|
||||
from .lfx_client import LfxClient
|
||||
|
||||
|
||||
def normalize_exported_flow(exported: dict, langflow_flow_id: str, name: str | None = None) -> dict:
|
||||
"""Reduce a Langflow FlowRead to the canonical fields we persist."""
|
||||
flow_name = name or exported.get("name") or f"langflow-{langflow_flow_id}"
|
||||
return {
|
||||
"id": exported.get("id", langflow_flow_id),
|
||||
"name": flow_name,
|
||||
"description": exported.get("description") or "",
|
||||
"endpoint_name": exported.get("endpoint_name"),
|
||||
"data": exported.get("data", {}),
|
||||
"tags": exported.get("tags", []),
|
||||
"webhook": exported.get("webhook", False),
|
||||
"is_component": exported.get("is_component", False),
|
||||
}
|
||||
|
||||
|
||||
async def pull_all_from_langflow(
|
||||
langflow: LangflowClient, flows_dir: str | Path
|
||||
) -> dict[str, list[str] | dict[str, str]]:
|
||||
"""Export every Langflow flow to its canonical file.
|
||||
|
||||
Returns {"pulled": [...stems], "errors": {langflow_id: reason}}.
|
||||
Per-flow failures are collected, never raised; listing failures raise
|
||||
(caller maps them to 503/502).
|
||||
"""
|
||||
flows_dir = Path(flows_dir)
|
||||
headers = await langflow.list_flows()
|
||||
pulled: list[str] = []
|
||||
errors: dict[str, str] = {}
|
||||
for header in headers:
|
||||
flow_id = str(header.get("id", ""))
|
||||
if not flow_id:
|
||||
continue
|
||||
try:
|
||||
exported = await langflow.export_flow(flow_id)
|
||||
doc = normalize_exported_flow(exported, flow_id)
|
||||
reuse, identical = export.find_reuse_path(flows_dir, doc)
|
||||
if identical and reuse is not None:
|
||||
pulled.append(reuse.stem)
|
||||
continue
|
||||
target = reuse or export.flow_path(flows_dir, export.filename_stem(doc["name"]))
|
||||
path = export.write_flow_bytes(target, doc)
|
||||
pulled.append(path.stem)
|
||||
except (LangflowError, OSError, ValueError) as exc:
|
||||
errors[flow_id] = str(exc)
|
||||
return {"pulled": sorted(pulled), "errors": errors}
|
||||
|
||||
|
||||
async def import_from_langflow(
|
||||
session: Session,
|
||||
langflow: LangflowClient,
|
||||
|
|
@ -28,18 +77,8 @@ async def import_from_langflow(
|
|||
truth; sync_document() derives the definition + version from it.
|
||||
"""
|
||||
exported = await langflow.export_flow(langflow_flow_id)
|
||||
flow_name = name or exported.get("name") or f"langflow-{langflow_flow_id}"
|
||||
langflow_json = {
|
||||
"id": exported.get("id", langflow_flow_id),
|
||||
"name": flow_name,
|
||||
"description": exported.get("description") or "",
|
||||
"endpoint_name": exported.get("endpoint_name"),
|
||||
"data": exported.get("data", {}),
|
||||
"tags": exported.get("tags", []),
|
||||
"webhook": exported.get("webhook", False),
|
||||
"is_component": exported.get("is_component", False),
|
||||
}
|
||||
path = export.write_flow_file(settings.flows_dir, langflow_json, name=flow_name)
|
||||
langflow_json = normalize_exported_flow(exported, langflow_flow_id, name=name)
|
||||
path = export.write_flow_file(settings.flows_dir, langflow_json, name=langflow_json["name"])
|
||||
definition, version, _action = await sync.sync_document(
|
||||
session,
|
||||
lfx,
|
||||
|
|
|
|||
|
|
@ -16,10 +16,17 @@ 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 filename_stem(name: str) -> str:
|
||||
"""Map a flow name to a filename stem, keeping the original name.
|
||||
|
||||
Only filesystem-reserved characters (``/ \\ : * ? " < > |`` and control
|
||||
chars) become ``-``; leading/trailing whitespace and trailing dots are
|
||||
stripped (Windows). Case, spaces, underscores and umlauts are preserved:
|
||||
``test_flow`` stays ``test_flow``.
|
||||
"""
|
||||
stem = re.sub(r'[\\/:*?"<>|\x00-\x1f]', "-", name).strip()
|
||||
stem = stem.rstrip(".")
|
||||
return stem or "flow"
|
||||
|
||||
|
||||
def canonical_bytes(doc: dict) -> bytes:
|
||||
|
|
@ -36,21 +43,55 @@ 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).
|
||||
def _read_doc(path: Path) -> dict | None:
|
||||
try:
|
||||
doc = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
return doc if isinstance(doc, dict) else None
|
||||
|
||||
The filename derives from `name` (or the document's own name). Returns
|
||||
the written path.
|
||||
|
||||
def find_reuse_path(flows_dir: str | Path, doc: dict) -> tuple[Path | None, bool]:
|
||||
"""Existing file to reuse for `doc`: (path, identical).
|
||||
|
||||
Matches identical content first (any filename), then the same flow name.
|
||||
Returns (None, False) when no candidate exists (caller writes a slug file).
|
||||
This keeps pulls idempotent: a flow seeded from `simple_rule.json` is
|
||||
updated in place instead of duplicated as `simple-rule.json`.
|
||||
"""
|
||||
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")
|
||||
if not flows_dir.is_dir():
|
||||
return None, False
|
||||
wanted_hash = content_hash(doc)
|
||||
wanted_name = doc.get("name")
|
||||
named: Path | None = None
|
||||
for path in iter_flow_files(flows_dir):
|
||||
existing = _read_doc(path)
|
||||
if existing is None:
|
||||
continue
|
||||
if content_hash(existing) == wanted_hash:
|
||||
return path, True
|
||||
if named is None and wanted_name and existing.get("name") == wanted_name:
|
||||
named = path
|
||||
return (named, False) if named is not None else (None, False)
|
||||
|
||||
|
||||
def write_flow_bytes(target: Path, doc: dict) -> Path:
|
||||
"""Write a flow document to an explicit path atomically (tmp + rename).
|
||||
|
||||
The file ends up world-readable (0o644): mkstemp creates 0o600, which
|
||||
would lock out the other containers (Langflow/LFX run as different
|
||||
users and crash on unreadable seed files).
|
||||
"""
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
fd, tmp_name = tempfile.mkstemp(
|
||||
dir=str(target.parent), prefix=f".{target.stem}-", suffix=".tmp"
|
||||
)
|
||||
try:
|
||||
with os.fdopen(fd, "wb") as tmp:
|
||||
tmp.write(canonical_bytes(doc))
|
||||
os.replace(tmp_name, target)
|
||||
os.chmod(target, 0o644)
|
||||
except BaseException:
|
||||
try:
|
||||
os.unlink(tmp_name)
|
||||
|
|
@ -60,6 +101,18 @@ def write_flow_file(flows_dir: str | Path, doc: dict, *, name: str | None = None
|
|||
return target
|
||||
|
||||
|
||||
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 = filename_stem(name or doc.get("name") or "flow")
|
||||
return write_flow_bytes(flow_path(flows_dir, stem), doc)
|
||||
|
||||
|
||||
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))
|
||||
|
|
|
|||
|
|
@ -127,13 +127,15 @@ class DecisionRead(BaseModel):
|
|||
|
||||
|
||||
class SyncReport(BaseModel):
|
||||
"""Result of reconciling backend/flows/*.json into DB + LFX.
|
||||
"""Result of the manual full pass: Langflow -> files -> DB + LFX.
|
||||
|
||||
Lists hold filename stems; errors maps filename to reason.
|
||||
Lists hold filename stems; errors maps filename (or Langflow id on
|
||||
pull failures) to reason.
|
||||
"""
|
||||
|
||||
added: list[str] = Field(default_factory=list)
|
||||
updated: list[str] = Field(default_factory=list)
|
||||
deleted: list[str] = Field(default_factory=list)
|
||||
skipped: list[str] = Field(default_factory=list)
|
||||
pulled: list[str] = Field(default_factory=list)
|
||||
errors: dict[str, str] = Field(default_factory=dict)
|
||||
|
|
|
|||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
747
backend/flows/test_flow.json
Normal file
747
backend/flows/test_flow.json
Normal file
File diff suppressed because one or more lines are too long
|
|
@ -18,8 +18,9 @@ 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 import sync
|
||||
from app.flows.deploy import pull_all_from_langflow
|
||||
from app.flows.langflow_client import LangflowClient
|
||||
from app.flows.lfx_client import LfxClient
|
||||
from app.flows.schemas import SyncReport
|
||||
from app.settings import settings
|
||||
|
|
@ -28,19 +29,7 @@ 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}
|
||||
return await pull_all_from_langflow(client, flows_dir)
|
||||
|
||||
|
||||
async def run_sync(flows_dir: Path) -> SyncReport:
|
||||
|
|
|
|||
|
|
@ -218,22 +218,36 @@ def override_langflow():
|
|||
|
||||
Usage: override_langflow(export_doc) -- the fake export_flow() returns the
|
||||
doc, or raises LangflowNotFound/Unavailable for status=404/0.
|
||||
Pass flows=[...] + docs={id: doc} for multi-flow pull tests, or
|
||||
list_status=0/500 to fail the listing. Default list is empty so
|
||||
file-only sync tests stay hermetic.
|
||||
"""
|
||||
from app.flows.api import get_langflow
|
||||
from app.flows.langflow_client import LangflowNotFound, LangflowUnavailable
|
||||
from app.flows.langflow_client import LangflowError, LangflowNotFound, LangflowUnavailable
|
||||
from app.main import app
|
||||
|
||||
def _override(export_doc=None, status=200):
|
||||
def _override(export_doc=None, status=200, flows=None, docs=None, list_status=200):
|
||||
if flows is None:
|
||||
flows = [{"id": "fake-1", "name": "fake"}] if export_doc is not None else []
|
||||
|
||||
class FakeLangflow:
|
||||
async def export_flow(self, flow_id):
|
||||
if status == 404:
|
||||
raise LangflowNotFound(f"flow {flow_id} not found")
|
||||
if status == 0:
|
||||
raise LangflowUnavailable("Langflow unreachable")
|
||||
if docs is not None:
|
||||
if flow_id in docs:
|
||||
return docs[flow_id]
|
||||
raise LangflowNotFound(f"flow {flow_id} not found")
|
||||
return export_doc
|
||||
|
||||
async def list_flows(self):
|
||||
return [{"id": "fake-1", "name": "fake"}]
|
||||
if list_status == 0:
|
||||
raise LangflowUnavailable("Langflow unreachable")
|
||||
if list_status == 500:
|
||||
raise LangflowError("Langflow error 500")
|
||||
return flows
|
||||
|
||||
app.dependency_overrides[get_langflow] = lambda: FakeLangflow()
|
||||
|
||||
|
|
|
|||
|
|
@ -5,11 +5,15 @@ import json
|
|||
from app.flows import export
|
||||
|
||||
|
||||
def test_slugify():
|
||||
assert export.slugify("Dachsanierung 200k") == "dachsanierung-200k"
|
||||
assert export.slugify(" A/B\\C ") == "a-b-c"
|
||||
assert export.slugify("simple_rule") == "simple-rule"
|
||||
assert export.slugify("") == "flow"
|
||||
def test_filename_stem_keeps_original_names():
|
||||
assert export.filename_stem("simple_rule") == "simple_rule"
|
||||
assert export.filename_stem("test_flow") == "test_flow"
|
||||
assert export.filename_stem("Gehaltserhöhung") == "Gehaltserhöhung"
|
||||
assert export.filename_stem("Dachsanierung 200k") == "Dachsanierung 200k"
|
||||
assert export.filename_stem(" A/B\\C ") == "A-B-C"
|
||||
assert export.filename_stem('a:b*c?d"e<f>g|h') == "a-b-c-d-e-f-g-h"
|
||||
assert export.filename_stem("trailing...") == "trailing"
|
||||
assert export.filename_stem("") == "flow"
|
||||
|
||||
|
||||
def test_canonical_bytes_stable_under_key_reorder():
|
||||
|
|
@ -22,7 +26,7 @@ def test_canonical_bytes_stable_under_key_reorder():
|
|||
def test_write_and_read_roundtrip_is_byte_stable(tmp_path):
|
||||
doc = {"name": "My Flow", "data": {"nodes": [], "edges": []}}
|
||||
path = export.write_flow_file(tmp_path, doc)
|
||||
assert path == tmp_path / "my-flow.json"
|
||||
assert path == tmp_path / "My Flow.json"
|
||||
assert json.loads(path.read_text()) == doc
|
||||
# Rewriting identical content is byte-identical (no phantom git diffs).
|
||||
before = path.read_bytes()
|
||||
|
|
@ -37,3 +41,42 @@ def test_iter_flow_files_only_top_level_json(tmp_path):
|
|||
sub.mkdir()
|
||||
(sub / "c.json").write_text("{}")
|
||||
assert export.iter_flow_files(tmp_path) == [tmp_path / "a.json"]
|
||||
|
||||
|
||||
def test_find_reuse_path_matches_identical_content_first(tmp_path):
|
||||
doc = {"name": "Simple Rule", "data": {"nodes": [], "edges": []}}
|
||||
other = tmp_path / "other-name.json"
|
||||
other.write_text(export.canonical_bytes(doc).decode())
|
||||
path, identical = export.find_reuse_path(tmp_path, dict(doc))
|
||||
assert (path, identical) == (other, True)
|
||||
|
||||
|
||||
def test_find_reuse_path_matches_same_name(tmp_path):
|
||||
old = {"name": "Simple Rule", "data": {"nodes": [], "edges": []}, "extra": 1}
|
||||
(tmp_path / "simple_rule.json").write_text(json.dumps(old))
|
||||
new = {"name": "Simple Rule", "data": {"nodes": [], "edges": []}}
|
||||
path, identical = export.find_reuse_path(tmp_path, new)
|
||||
assert (path, identical) == (tmp_path / "simple_rule.json", False)
|
||||
|
||||
|
||||
def test_find_reuse_path_no_match_and_missing_dir(tmp_path):
|
||||
(tmp_path / "a.json").write_text(json.dumps({"name": "Other"}))
|
||||
assert export.find_reuse_path(tmp_path, {"name": "Neu"}) == (None, False)
|
||||
assert export.find_reuse_path(tmp_path / "gibts-nicht", {"name": "Neu"}) == (None, False)
|
||||
|
||||
|
||||
def test_find_reuse_path_skips_unreadable_files(tmp_path):
|
||||
(tmp_path / "kaputt.json").write_text("{kein json")
|
||||
assert export.find_reuse_path(tmp_path, {"name": "Neu"}) == (None, False)
|
||||
|
||||
|
||||
def test_write_flow_file_is_world_readable(tmp_path):
|
||||
import os
|
||||
import stat
|
||||
|
||||
doc = {"name": "Lesbar", "data": {"nodes": [], "edges": []}}
|
||||
path = export.write_flow_file(tmp_path, doc)
|
||||
mode = stat.S_IMODE(os.stat(path).st_mode)
|
||||
# Read bits for user/group/other (mkstemp defaults to 0o600; on Windows
|
||||
# chmod maps to 0o666, so only assert the read bits, not the exact mode).
|
||||
assert mode & 0o444 == 0o444, f"expected world-readable, got {oct(mode)}"
|
||||
|
|
|
|||
|
|
@ -118,20 +118,127 @@ async def test_sync_rename_updates_definition_name(session, lfx_server, tmp_path
|
|||
assert definition.name == "umbenannt"
|
||||
|
||||
|
||||
async def test_sync_endpoint_returns_report(client, lfx_server, tmp_path, monkeypatch, login):
|
||||
async def test_sync_endpoint_returns_report(
|
||||
client, lfx_server, tmp_path, monkeypatch, login, override_langflow
|
||||
):
|
||||
await login()
|
||||
override_langflow() # empty Langflow: file-only pass
|
||||
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
||||
_seed_dir(tmp_path)
|
||||
response = await client.post("/api/flows/sync")
|
||||
assert response.status_code == 200, response.text
|
||||
body = response.json()
|
||||
assert body["added"] == ["simple_rule"]
|
||||
assert body["pulled"] == []
|
||||
assert body["errors"] == {}
|
||||
|
||||
response = await client.post("/api/flows/sync")
|
||||
assert response.json()["skipped"] == ["simple_rule"]
|
||||
|
||||
|
||||
def _langflow_doc(name: str, langflow_id: str) -> dict:
|
||||
doc = load_fixture("simple_rule.json")
|
||||
return {
|
||||
"id": langflow_id,
|
||||
"name": name,
|
||||
"description": "aus Langflow",
|
||||
"endpoint_name": None,
|
||||
"data": doc["data"],
|
||||
"tags": [],
|
||||
"webhook": False,
|
||||
"is_component": False,
|
||||
}
|
||||
|
||||
|
||||
async def test_sync_endpoint_pulls_from_langflow_then_syncs(
|
||||
client, lfx_server, tmp_path, monkeypatch, login, override_langflow
|
||||
):
|
||||
await login()
|
||||
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
||||
langflow_id = "22222222-3333-4444-8555-666666666666"
|
||||
override_langflow(
|
||||
flows=[{"id": langflow_id, "name": "Gehaltserhöhung"}],
|
||||
docs={langflow_id: _langflow_doc("Gehaltserhöhung", langflow_id)},
|
||||
)
|
||||
response = await client.post("/api/flows/sync")
|
||||
assert response.status_code == 200, response.text
|
||||
body = response.json()
|
||||
assert body["pulled"] == ["Gehaltserhöhung"]
|
||||
assert body["added"] == ["Gehaltserhöhung"]
|
||||
assert body["errors"] == {}
|
||||
assert (tmp_path / "Gehaltserhöhung.json").is_file()
|
||||
|
||||
# Second pass: same content pulls again but syncs nothing new.
|
||||
response = await client.post("/api/flows/sync")
|
||||
assert response.status_code == 200, response.text
|
||||
body = response.json()
|
||||
assert body["pulled"] == ["Gehaltserhöhung"]
|
||||
assert body["skipped"] == ["Gehaltserhöhung"]
|
||||
|
||||
|
||||
async def test_sync_endpoint_pull_reuses_same_named_file(
|
||||
client, lfx_server, tmp_path, monkeypatch, login, override_langflow
|
||||
):
|
||||
"""A pulled flow seeded from simple_rule.json updates it in place
|
||||
instead of duplicating it as simple-rule.json (slug drift)."""
|
||||
await login()
|
||||
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
||||
_seed_dir(tmp_path) # fixture shape, name "simple_rule"
|
||||
langflow_id = "44444444-5555-4666-8777-888888888888"
|
||||
override_langflow(
|
||||
flows=[{"id": langflow_id, "name": "simple_rule"}],
|
||||
docs={langflow_id: _langflow_doc("simple_rule", langflow_id)},
|
||||
)
|
||||
response = await client.post("/api/flows/sync")
|
||||
assert response.status_code == 200, response.text
|
||||
body = response.json()
|
||||
assert body["pulled"] == ["simple_rule"]
|
||||
assert body["added"] == ["simple_rule"]
|
||||
assert body["errors"] == {}
|
||||
assert not (tmp_path / "simple-rule.json").exists()
|
||||
|
||||
response = await client.post("/api/flows/sync")
|
||||
assert response.json()["skipped"] == ["simple_rule"]
|
||||
|
||||
|
||||
async def test_sync_endpoint_pull_error_continues(
|
||||
client, lfx_server, tmp_path, monkeypatch, login, override_langflow
|
||||
):
|
||||
await login()
|
||||
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
||||
good_id = "33333333-4444-4555-8666-777777777777"
|
||||
override_langflow(
|
||||
flows=[{"id": good_id, "name": "Gut"}, {"id": "kaputt-id", "name": "Kaputt"}],
|
||||
docs={good_id: _langflow_doc("Gut", good_id)},
|
||||
)
|
||||
response = await client.post("/api/flows/sync")
|
||||
assert response.status_code == 200, response.text
|
||||
body = response.json()
|
||||
assert body["pulled"] == ["Gut"]
|
||||
assert body["added"] == ["Gut"]
|
||||
assert "kaputt-id" in body["errors"]
|
||||
|
||||
|
||||
async def test_sync_endpoint_langflow_unreachable(
|
||||
client, lfx_server, tmp_path, monkeypatch, login, override_langflow
|
||||
):
|
||||
await login()
|
||||
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
||||
override_langflow(flows=[], list_status=0)
|
||||
response = await client.post("/api/flows/sync")
|
||||
assert response.status_code == 503
|
||||
|
||||
|
||||
async def test_sync_endpoint_langflow_error(
|
||||
client, lfx_server, tmp_path, monkeypatch, login, override_langflow
|
||||
):
|
||||
await login()
|
||||
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
||||
override_langflow(flows=[], list_status=500)
|
||||
response = await client.post("/api/flows/sync")
|
||||
assert response.status_code == 502
|
||||
|
||||
|
||||
async def test_watch_forever_triggers_on_file_change(tmp_path):
|
||||
import contextlib
|
||||
|
||||
|
|
|
|||
|
|
@ -66,6 +66,7 @@ export interface SyncReport {
|
|||
updated: string[];
|
||||
deleted: string[];
|
||||
skipped: string[];
|
||||
pulled: string[];
|
||||
errors: Record<string, string>;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ import {
|
|||
type FlowDefinition,
|
||||
type FlowVersion,
|
||||
} from "../api";
|
||||
import FlowHelp from "./FlowHelp";
|
||||
import StatusBadge from "./StatusBadge";
|
||||
|
||||
export default function AdminTab() {
|
||||
|
|
@ -89,6 +90,7 @@ export default function AdminTab() {
|
|||
try {
|
||||
const r = await api.syncFlows();
|
||||
const parts = [
|
||||
`Langflow: ${r.pulled.length}`,
|
||||
`+${r.added.length}`,
|
||||
`~${r.updated.length}`,
|
||||
`-${r.deleted.length}`,
|
||||
|
|
@ -147,6 +149,7 @@ export default function AdminTab() {
|
|||
</Button>
|
||||
</Group>
|
||||
</Card>
|
||||
<FlowHelp />
|
||||
<Card withBorder padding="md" mb="md">
|
||||
<Group>
|
||||
<Button
|
||||
|
|
@ -154,9 +157,13 @@ export default function AdminTab() {
|
|||
leftSection={<IconRefresh size={16} />}
|
||||
onClick={() => void onSync()}
|
||||
>
|
||||
Aus backend/flows synchronisieren
|
||||
Aus Langflow synchronisieren
|
||||
</Button>
|
||||
</Group>
|
||||
<Text size="sm" c="dimmed" mt="xs">
|
||||
Holt alle Flows aus Langflow nach backend/flows und deployed sie
|
||||
danach automatisch.
|
||||
</Text>
|
||||
</Card>
|
||||
{flows.length === 0 ? (
|
||||
<Text c="dimmed">Keine Flows vorhanden.</Text>
|
||||
|
|
|
|||
93
frontend/src/components/FlowHelp.tsx
Normal file
93
frontend/src/components/FlowHelp.tsx
Normal file
|
|
@ -0,0 +1,93 @@
|
|||
import { useState } from "react";
|
||||
import {
|
||||
Anchor,
|
||||
Button,
|
||||
Card,
|
||||
List,
|
||||
Text,
|
||||
Title,
|
||||
} from "@mantine/core";
|
||||
import { IconChevronDown, IconChevronUp, IconExternalLink } from "@tabler/icons-react";
|
||||
import { LANGFLOW_URL } from "../api";
|
||||
|
||||
export default function FlowHelp() {
|
||||
const [open, setOpen] = useState(false);
|
||||
|
||||
return (
|
||||
<Card withBorder padding="md" mb="md">
|
||||
<Button
|
||||
variant="subtle"
|
||||
leftSection={open ? <IconChevronUp size={16} /> : <IconChevronDown size={16} />}
|
||||
onClick={() => setOpen((v) => !v)}
|
||||
>
|
||||
{open ? "Anleitung verbergen" : "Anleitung: Neuer Flow in Langflow"}
|
||||
</Button>
|
||||
{open && (
|
||||
<div>
|
||||
<Title order={4} mt="sm" mb="xs">
|
||||
So kommt ein neuer Flow in die Webapp
|
||||
</Title>
|
||||
<List size="sm" spacing="xs">
|
||||
<List.Item>
|
||||
<Text component="span" fw={600}>
|
||||
1. In Langflow entwerfen und speichern:{" "}
|
||||
</Text>
|
||||
<Anchor href={LANGFLOW_URL} target="_blank" rel="noreferrer">
|
||||
Langflow öffnen <IconExternalLink size={14} />
|
||||
</Anchor>
|
||||
. Die bestehenden Flows sind dort bereits geladen. Name vergeben, speichern.
|
||||
</List.Item>
|
||||
<List.Item>
|
||||
<Text component="span" fw={600}>
|
||||
2. Sync-Button unten klicken:{" "}
|
||||
</Text>
|
||||
holt alle Flows aus Langflow nach{" "}
|
||||
<Text component="span" ff="monospace">
|
||||
backend/flows/*.json
|
||||
</Text>{" "}
|
||||
und deployed sie danach automatisch (neue Version nur bei geändertem Inhalt,
|
||||
sonst übersprungen).
|
||||
</List.Item>
|
||||
<List.Item>
|
||||
<Text component="span" fw={600}>
|
||||
3. Prüfen:{" "}
|
||||
</Text>
|
||||
Tabelle unten zeigt Version, Autor und Status. Danach erscheint der Flow
|
||||
automatisch auf der Seite Workflows mit seinem Formular. Flows ohne
|
||||
Formular-Knoten zeigen ein Freitext-Feld.
|
||||
</List.Item>
|
||||
</List>
|
||||
<Title order={5} mt="md" mb="xs">
|
||||
Regeln
|
||||
</Title>
|
||||
<List size="sm" spacing="xs">
|
||||
<List.Item>
|
||||
Die Datei heißt wie der Flow in Langflow (z. B. test_flow.json).
|
||||
Umbenennen in Langflow erzeugt eine neue Datei; die alte bleibt
|
||||
bestehen und muss bei Bedarf gelöscht werden.
|
||||
</List.Item>
|
||||
<List.Item>
|
||||
Eine gelöschte Datei löscht den Flow inklusive seiner Läufe
|
||||
(Datei bleibt in Git erhalten).
|
||||
</List.Item>
|
||||
<List.Item>
|
||||
Fehler stehen in der Sync-Meldung (z. B. ungültiges Flow-JSON).
|
||||
Der automatische Beobachter im Hintergrund syncet nur Dateien,
|
||||
Langflow holt nur dieser Button.
|
||||
</List.Item>
|
||||
</List>
|
||||
<Title order={5} mt="md" mb="xs">
|
||||
Formular-Felder
|
||||
</Title>
|
||||
<Text size="sm">
|
||||
Für ein Web-Formular braucht der Flow einen CustomInputForm-Knoten: jeder
|
||||
Eintrag dort ist ein Formularfeld (Name = Feld-Key, Anzeigename = Label).
|
||||
Die Verbindung ChatInput nach form_values bitte behalten — darüber kommen
|
||||
die ausgefüllten Werte an. Unbekannte Feldtypen fallen automatisch auf ein
|
||||
Textfeld zurück.
|
||||
</Text>
|
||||
</div>
|
||||
)}
|
||||
</Card>
|
||||
);
|
||||
}
|
||||
9
lfx-entrypoint.sh
Normal file
9
lfx-entrypoint.sh
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
#!/bin/sh
|
||||
# LFX entrypoint (runs as root, drops privileges for the server).
|
||||
# The named volume arrives root-owned (Docker creates it as root and the
|
||||
# image USER cannot chown it) — without this every flow upload fails with
|
||||
# "Permission denied" and synced flows stay version-less (invisible in UI).
|
||||
set -e
|
||||
|
||||
chown -R lfx:lfx /flows
|
||||
exec gosu lfx "$@"
|
||||
Loading…
Add table
Add a link
Reference in a new issue