Compare commits
5 commits
74831483eb
...
9acdbaa453
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9acdbaa453 | ||
|
|
94f353d773 | ||
|
|
338cdcf108 | ||
|
|
c65e3ed332 | ||
|
|
7d5ade6b6a |
37 changed files with 4696 additions and 115 deletions
|
|
@ -8,7 +8,7 @@ ENV PYTHONUNBUFFERED=1 \
|
|||
PIP_NO_CACHE_DIR=1
|
||||
|
||||
# lfx = flow runtime; requests = convenience for custom-component code.
|
||||
RUN pip install "lfx==1.12.0" requests
|
||||
RUN pip install lfx requests
|
||||
|
||||
RUN mkdir -p /flows
|
||||
VOLUME /flows
|
||||
|
|
|
|||
|
|
@ -19,26 +19,8 @@ services:
|
|||
- "8000:8000" # FastAPI
|
||||
- "3000:3000" # Vite
|
||||
|
||||
db:
|
||||
image: postgres:16-trixie
|
||||
env_file:
|
||||
- ../.env
|
||||
volumes:
|
||||
- postgres-data:/var/lib/postgresql/data
|
||||
- ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql:ro
|
||||
ports:
|
||||
- "5432:5432"
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "pg_isready -U postgres"]
|
||||
interval: 5s
|
||||
timeout: 3s
|
||||
retries: 10
|
||||
|
||||
langflow:
|
||||
image: langflowai/langflow:1.12.0
|
||||
depends_on:
|
||||
db:
|
||||
condition: service_healthy
|
||||
image: langflowai/langflow:latest
|
||||
env_file:
|
||||
- ../.env
|
||||
environment:
|
||||
|
|
@ -49,6 +31,9 @@ services:
|
|||
PORT: "7860"
|
||||
LANGFLOW_AUTO_LOGIN: "false"
|
||||
LANGFLOW_API_KEY_SOURCE: env
|
||||
# SQLite file inside the langflow-data volume (absolute path required
|
||||
# by Langflow); overrides any postgres URL in ../.env.
|
||||
LANGFLOW_DATABASE_URL: sqlite:////app/langflow/langflow.db
|
||||
volumes:
|
||||
- langflow-data:/app/langflow
|
||||
ports:
|
||||
|
|
@ -85,9 +70,23 @@ services:
|
|||
retries: 12
|
||||
start_period: 30s
|
||||
|
||||
# Optional nightly auto-updates of :latest images. Not part of the default
|
||||
# dev startup (no container updates needed while iterating).
|
||||
watchtower:
|
||||
image: nickfedor/watchtower:latest
|
||||
profiles: ["updates"]
|
||||
volumes:
|
||||
- /var/run/docker.sock:/var/run/docker.sock
|
||||
environment:
|
||||
WATCHTOWER_LABEL_ENABLE: "true"
|
||||
WATCHTOWER_SCHEDULE: "0 0 4 * * *"
|
||||
WATCHTOWER_CLEANUP: "true"
|
||||
WATCHTOWER_ROLLING_RESTART: "true"
|
||||
labels:
|
||||
com.centurylinklabs.watchtower.enable: "true"
|
||||
|
||||
volumes:
|
||||
kirche-einfach-venv:
|
||||
kirche-einfach-node-modules:
|
||||
postgres-data:
|
||||
langflow-data:
|
||||
lfx-data:
|
||||
lfx-data:
|
||||
|
|
@ -1,7 +0,0 @@
|
|||
-- Runs once on first init of the postgres-data volume.
|
||||
-- The main app database ($POSTGRES_DB) is created by the image itself.
|
||||
-- Langflow (authoring UI) gets its own database on the same instance.
|
||||
CREATE DATABASE langflow;
|
||||
|
||||
-- Isolated database for the pytest suite (migrated via Alembic in conftest).
|
||||
CREATE DATABASE kirche_einfach_test;
|
||||
|
|
@ -13,7 +13,7 @@ echo "--- Installing Node dependencies ---"
|
|||
cd /workspaces/kirche-einfach/frontend
|
||||
npm install
|
||||
|
||||
# Compose services (db, langflow, lfx) must be up; fail loudly otherwise.
|
||||
# Compose services (langflow, lfx) must be up; fail loudly otherwise.
|
||||
echo "--- Waiting for compose services ---"
|
||||
if [ ! -f /workspaces/kirche-einfach/.env ]; then
|
||||
echo "ERROR: /workspaces/kirche-einfach/.env is missing. Copy .env.example to .env and fill in secrets." >&2
|
||||
|
|
@ -24,7 +24,7 @@ import socket
|
|||
import sys
|
||||
import time
|
||||
|
||||
targets = [("db", 5432), ("langflow", 7860), ("lfx", 6004)]
|
||||
targets = [("langflow", 7860), ("lfx", 6004)]
|
||||
deadline = time.time() + 180
|
||||
pending = dict(targets)
|
||||
while pending and time.time() < deadline:
|
||||
|
|
@ -45,6 +45,14 @@ if pending:
|
|||
sys.exit(f"ERROR: compose services not ready: {sorted(pending)}")
|
||||
EOF
|
||||
|
||||
# Migrate the SQLite schema and reconcile backend/flows into the DB + LFX.
|
||||
echo "--- Migrating database ---"
|
||||
cd /workspaces/kirche-einfach/backend
|
||||
uv run alembic upgrade head
|
||||
|
||||
echo "--- Syncing canonical flows ---"
|
||||
uv run python scripts/sync_flows.py --sync || echo "WARN: flow sync failed (check backend/flows and LFX)"
|
||||
|
||||
echo ""
|
||||
echo "=== Setup complete ==="
|
||||
echo "Forgejo username: $FORGEJO_USERNAME"
|
||||
|
|
|
|||
18
.env.example
18
.env.example
|
|
@ -1,12 +1,14 @@
|
|||
# Copy to `.env` and fill in values. `.env` is gitignored.
|
||||
FORGEJO_TOKEN=
|
||||
|
||||
# --- databases (compose service `db`) ---
|
||||
POSTGRES_USER=kirche
|
||||
POSTGRES_PASSWORD=
|
||||
POSTGRES_DB=kirche_einfach
|
||||
DATABASE_URL=postgresql+psycopg://kirche:<password>@db:5432/kirche_einfach
|
||||
TEST_DATABASE_URL=postgresql+psycopg://kirche:<password>@db:5432/kirche_einfach_test
|
||||
# --- app database (SQLite, default if unset) ---
|
||||
# Override only if you want a different engine/URL.
|
||||
# DATABASE_URL=sqlite:///backend/data/kirche.db
|
||||
|
||||
# --- canonical flow definitions ---
|
||||
# FLOWS_DIR=backend/flows
|
||||
# Auto-reconcile backend/flows into DB + LFX while the API runs.
|
||||
# FLOWS_WATCH_ENABLED=true
|
||||
|
||||
# --- langflow (authoring, dev only) ---
|
||||
LANGFLOW_URL=http://langflow:7860
|
||||
|
|
@ -15,7 +17,7 @@ LANGFLOW_SUPERUSER_PASSWORD=
|
|||
LANGFLOW_SECRET_KEY=
|
||||
# Single static key; validated against env (LANGFLOW_API_KEY_SOURCE=env), superuser scope.
|
||||
LANGFLOW_API_KEY=sk-
|
||||
LANGFLOW_DATABASE_URL=postgresql://kirche:<password>@db:5432/langflow
|
||||
# Langflow stores flows in its own SQLite file when unset.
|
||||
|
||||
# --- lfx (flow runtime) ---
|
||||
LFX_URL=http://lfx:6004
|
||||
|
|
@ -25,4 +27,4 @@ LFX_API_KEY=lfx-
|
|||
CALLBACK_BASE_URL=http://app:8000
|
||||
|
||||
# --- LLM (optional; tests mock the provider without it) ---
|
||||
# LLM_API_KEY=
|
||||
# LLM_API_KEY=
|
||||
6
.gitignore
vendored
6
.gitignore
vendored
|
|
@ -14,6 +14,12 @@ dist/
|
|||
.env
|
||||
.env.local
|
||||
|
||||
# Runtime SQLite databases
|
||||
backend/data/
|
||||
*.db
|
||||
*.db-shm
|
||||
*.db-wal
|
||||
|
||||
# IDE
|
||||
.vscode/
|
||||
.idea/
|
||||
|
|
|
|||
41
AGENTS.md
41
AGENTS.md
|
|
@ -1,5 +1,44 @@
|
|||
# Agents
|
||||
|
||||
Kirche Einfach — AI-assisted church administration. FastAPI backend (`backend/`, Python 3.12, SQLModel, SQLite) + React 19/Vite frontend (`frontend/`). Flow definitions live as JSON files in `backend/flows/` (source of truth); the DB is derived by the sync engine.
|
||||
|
||||
## Commands
|
||||
|
||||
Backend (run from `backend/`):
|
||||
- `uv sync` — install dependencies
|
||||
- `uv run pytest` — full test suite (throwaway SQLite file, no external DB needed)
|
||||
- `uv run alembic upgrade head` — migrate the dev database (`backend/data/`, gitignored)
|
||||
- `uv run python scripts/sync_flows.py --sync` — reconcile `backend/flows/` into DB + LFX
|
||||
- `uv run uvicorn app.main:app --port 8000` — run the API (paths resolve from repo root, CWD-independent)
|
||||
|
||||
Frontend (run from `frontend/`): `npm install`, `npm run dev`, `npm run build` (typecheck + build).
|
||||
|
||||
## Definition of done
|
||||
|
||||
- New behavior has tests; `uv run pytest` is green before reporting completion.
|
||||
- `npm run build` is green when frontend code changed.
|
||||
- Schema changes ship as a new Alembic revision — never edit an applied migration.
|
||||
|
||||
## Architecture rules
|
||||
|
||||
- `backend/flows/*.json` is the source of truth (one Langflow export per file). Never write flow JSON anywhere else.
|
||||
- Only `backend/app/flows/sync.py` reconciles files into the DB. CLI, `POST /api/flows/sync`, and the embedded watcher must call it — never duplicate its logic.
|
||||
- All DB access goes through `backend/app/flows/store.py`. Keep API handlers thin.
|
||||
- Talk to LFX only via `LfxClient`, to Langflow only via `LangflowClient`.
|
||||
- Flow JSON must stay a valid Langflow export (`store` requires `data.nodes`/`data.edges`).
|
||||
|
||||
```python
|
||||
from app.flows.sync import sync_from_directory
|
||||
report = await sync_from_directory(session, lfx, settings.flows_dir)
|
||||
```
|
||||
|
||||
## Boundaries
|
||||
|
||||
- Never commit secrets. `.env` is gitignored; only `.env.example` is tracked.
|
||||
- Never edit `backend/data/*` (runtime SQLite files) or `__pycache__`.
|
||||
- Never add a second file-watcher or a second sync path.
|
||||
- Deleting `backend/flows/<name>.json` deletes the flow including its runs from the DB (by design).
|
||||
|
||||
## Git
|
||||
|
||||
When committing: `git add -A && git commit -m "<message>"`. When pushing: `git push` (branch is tracking `origin/main`).
|
||||
When committing: `git add -A && git commit -m "<message>"` (conventional commits: `feat:`, `fix:`, `chore:`, `docs:`). When pushing: `git push` (branch is tracking `origin/main`).
|
||||
|
|
|
|||
104
README.md
104
README.md
|
|
@ -1,17 +1,103 @@
|
|||
# Kirche Einfach
|
||||
|
||||
KI-gestützte Verwaltungsplattform für Kirche.
|
||||
AI-assisted administration platform for church organizations.
|
||||
|
||||
## Stack
|
||||
Staff describe a request in plain words. The platform guides them through the
|
||||
process, checks the rules automatically, and asks a human only when something
|
||||
needs judgment. Every step is documented and traceable.
|
||||
|
||||
- **Backend:** Python 3.12, FastAPI, Django, uv
|
||||
- **Frontend:** React 19, TypeScript, Vite
|
||||
- **Dev:** Devcontainer, opencode
|
||||
## Features
|
||||
|
||||
- **Visual flow authoring** — design decision flows in Langflow, then bring
|
||||
them into the platform.
|
||||
- **Flows as files** — each flow lives as one JSON file under
|
||||
`backend/flows/`. The directory is the source of truth: committed to git,
|
||||
reviewable, diffable.
|
||||
- **Automatic sync** — the API watches `backend/flows/` and reconciles every
|
||||
change into the database and the runtime: new files become flows, edits
|
||||
become new versions, deleted files remove the flow (including its runs).
|
||||
Same on every startup; also triggerable via `POST /api/flows/sync` or
|
||||
`uv run python scripts/sync_flows.py --sync`.
|
||||
- **Versioned runtime** — every change creates an immutable version; versions
|
||||
deploy to the lightweight LFX runtime and can run synchronously or streamed.
|
||||
- **Approvals** — a flow can pause and ask a human for a decision. Approving
|
||||
can automatically start a follow-up ("continuation") flow.
|
||||
- **Web UI** — three tabs: *Vorgänge* (flows and runs), *Freigaben*
|
||||
(approvals), *Administration* (import, sync, deploy).
|
||||
|
||||
## Infrastructure
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
subgraph Repo["Git repo — source of truth"]
|
||||
FLOWS["backend/flows/*.json<br/>one Langflow export per flow"]
|
||||
end
|
||||
subgraph Stack["Compose stack (dev + prod)"]
|
||||
UI["frontend — React UI (3000)"]
|
||||
API["app — FastAPI (8000)<br/>+ embedded flow-watcher"]
|
||||
DB[("app.db — SQLite file")]
|
||||
LF["langflow — authoring UI (7860)<br/>langflow.db — SQLite file"]
|
||||
LFX["lfx — flow runtime (6004)"]
|
||||
WT["watchtower — nightly image updates"]
|
||||
end
|
||||
UI --> API
|
||||
FLOWS -->|watch + sync| API
|
||||
API -->|definitions · versions · runs · approvals| DB
|
||||
API -->|pull / export| LF
|
||||
API -->|upload · run · unregister| LFX
|
||||
LFX -->|approval callbacks| API
|
||||
WT -.->|updates :latest images| LF
|
||||
```
|
||||
|
||||
| Service | Purpose | URL |
|
||||
| -------------- | ------------------------------------ | -------------------------- |
|
||||
| React frontend | Web UI (Vite) | http://localhost:3000 |
|
||||
| FastAPI backend| API + flow orchestration | http://localhost:8000 |
|
||||
| API docs | Swagger UI | http://localhost:8000/api/docs |
|
||||
| Langflow | Visual flow authoring | http://localhost:7860 |
|
||||
| LFX | Flow runtime (internal, no direct UI)| localhost:6004 |
|
||||
| SQLite files | `backend/data/kirche.db` (app), Langflow keeps its own file | — |
|
||||
|
||||
Why SQLite: a flow run needs only a handful of single-row statements, and
|
||||
Langflow itself runs on SQLite by default — so the whole stack works without
|
||||
a database server. Postgres can return later by changing `DATABASE_URL`
|
||||
(migrations stay portable).
|
||||
|
||||
## Getting Started
|
||||
|
||||
Open in VS Code → "Reopen in Container".
|
||||
1. Copy `.env.example` to `.env` and fill in the secrets.
|
||||
2. Open the repo in VS Code → "Reopen in Container". Setup installs
|
||||
dependencies, migrates the database, and syncs `backend/flows/`.
|
||||
3. Run the backend from `backend/`: `uv run uvicorn app.main:app --port 8000`.
|
||||
Run the frontend from `frontend/`: `npm run dev`.
|
||||
|
||||
Backend: `http://localhost:8000`
|
||||
Frontend: `http://localhost:3000`
|
||||
API Docs: `http://localhost:8000/api/docs`
|
||||
Production (self-hosted, internal use — no TLS/reverse proxy included):
|
||||
|
||||
```sh
|
||||
docker compose -f compose.prod.yml up -d --build
|
||||
```
|
||||
|
||||
## Development
|
||||
|
||||
Backend (from `backend/`):
|
||||
|
||||
- `uv run pytest` — full test suite (throwaway SQLite file, no external DB).
|
||||
- `uv run alembic upgrade head` — migrate the dev database.
|
||||
- `uv run python scripts/sync_flows.py --sync` — reconcile `backend/flows/`.
|
||||
- `uv run python scripts/sync_flows.py --pull` — export all Langflow flows
|
||||
into `backend/flows/` first.
|
||||
- `uv run python scripts/deploy_flow.py --langflow-id <uuid>` — import one
|
||||
Langflow flow via the API.
|
||||
- `uv run python scripts/make_flow_fixtures.py` — regenerate the bundled
|
||||
flows (writes test fixtures and `backend/flows/`).
|
||||
- `uv run python scripts/smoke_e2e.py` — manual end-to-end smoke.
|
||||
|
||||
Frontend (from `frontend/`): `npm install`, `npm run dev`, `npm run build`.
|
||||
|
||||
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).
|
||||
|
||||
Container updates: registry images (`langflowai/langflow:latest`)
|
||||
auto-update nightly via Watchtower (`nickfedor/watchtower`). Locally built
|
||||
images (`app`, `lfx`, `frontend`) update with `docker compose build --pull`.
|
||||
|
|
|
|||
21
backend/Dockerfile
Normal file
21
backend/Dockerfile
Normal file
|
|
@ -0,0 +1,21 @@
|
|||
# Production image for the FastAPI backend.
|
||||
FROM python:3.12-slim
|
||||
|
||||
ENV PYTHONUNBUFFERED=1 \
|
||||
PIP_NO_CACHE_DIR=1
|
||||
|
||||
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
|
||||
|
||||
WORKDIR /srv/backend
|
||||
COPY pyproject.toml alembic.ini ./
|
||||
COPY app ./app
|
||||
COPY migrations ./migrations
|
||||
COPY scripts ./scripts
|
||||
# No lockfile is committed; dependencies resolve at build time.
|
||||
RUN uv sync --no-dev
|
||||
|
||||
ENV FLOWS_WATCH_ENABLED=true
|
||||
EXPOSE 8000
|
||||
|
||||
# Migrate on start, then serve (watcher reconciles backend/flows automatically).
|
||||
CMD ["sh", "-c", ".venv/bin/alembic upgrade head && exec .venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000"]
|
||||
|
|
@ -1,6 +1,7 @@
|
|||
"""Database engine, sessions, and FastAPI dependency."""
|
||||
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import Engine
|
||||
from sqlmodel import Session, SQLModel, create_engine
|
||||
|
|
@ -10,11 +11,23 @@ from .settings import settings
|
|||
_engine: Engine | None = None
|
||||
|
||||
|
||||
def _engine_kwargs(url: str) -> dict:
|
||||
if url.startswith("sqlite"):
|
||||
# FastAPI runs sync endpoints in a threadpool; connections are shared
|
||||
# across threads, so defer the same-thread check. File-based SQLite
|
||||
# needs an existing parent dir (SQLite won't create it).
|
||||
return {"connect_args": {"check_same_thread": False}}
|
||||
return {}
|
||||
|
||||
|
||||
def get_engine(url: str | None = None) -> Engine:
|
||||
"""Process-wide engine singleton (override URL in tests via settings)."""
|
||||
global _engine
|
||||
if _engine is None:
|
||||
_engine = create_engine(url or settings.database_url)
|
||||
url = url or settings.database_url
|
||||
if url.startswith("sqlite:///") and not url.startswith("sqlite:///:memory:"):
|
||||
Path(url.removeprefix("sqlite:///")).parent.mkdir(parents=True, exist_ok=True)
|
||||
_engine = create_engine(url, **_engine_kwargs(url))
|
||||
return _engine
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ from __future__ import annotations
|
|||
|
||||
import json
|
||||
from collections.abc import AsyncIterator, Iterator
|
||||
from pathlib import Path
|
||||
from uuid import UUID
|
||||
|
||||
import httpx
|
||||
|
|
@ -16,6 +17,7 @@ from app.settings import settings
|
|||
|
||||
from . import runner, store
|
||||
from .deploy import import_from_langflow
|
||||
from .sync import lfx_id_for, sync_from_directory
|
||||
from .langflow_client import (
|
||||
LangflowClient,
|
||||
LangflowError,
|
||||
|
|
@ -44,6 +46,7 @@ from .schemas import (
|
|||
FlowVersionRead,
|
||||
RunRead,
|
||||
RunRequest,
|
||||
SyncReport,
|
||||
)
|
||||
|
||||
flows_router = APIRouter(prefix="/api/flows", tags=["flows"])
|
||||
|
|
@ -70,6 +73,7 @@ def _read_definition(d) -> FlowDefinitionRead:
|
|||
name=d.name,
|
||||
description=d.description,
|
||||
status=d.status,
|
||||
source_path=d.source_path,
|
||||
continuation_flow_id=d.continuation_flow_id,
|
||||
created_at=d.created_at,
|
||||
)
|
||||
|
|
@ -134,6 +138,18 @@ def list_flows(session: Session = Depends(get_db)):
|
|||
return [_read_definition(d) for d in store.list_definitions(session)]
|
||||
|
||||
|
||||
@flows_router.post("/sync", response_model=SyncReport)
|
||||
async def sync_flows(
|
||||
session: Session = Depends(get_db),
|
||||
lfx: LfxClient = Depends(get_lfx),
|
||||
):
|
||||
"""Reconcile backend/flows/*.json into the DB + LFX (add/update/delete)."""
|
||||
try:
|
||||
return await sync_from_directory(session, lfx, Path(settings.flows_dir))
|
||||
except FileNotFoundError as exc:
|
||||
raise HTTPException(status_code=500, detail=str(exc)) from exc
|
||||
|
||||
|
||||
@flows_router.get("/{flow_id}", response_model=FlowDefinitionRead)
|
||||
def get_flow(flow_id: UUID, session: Session = Depends(get_db)):
|
||||
try:
|
||||
|
|
@ -178,8 +194,12 @@ async def deploy_version(
|
|||
except store.NotFoundError as exc:
|
||||
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
||||
try:
|
||||
# Deterministic LFX id: same (definition, version) always maps to the
|
||||
# same registration, no matter which path deployed it.
|
||||
uploaded = await lfx.upload_flow(
|
||||
version.langflow_json, flow_id=str(version.id), replace=True
|
||||
version.langflow_json,
|
||||
flow_id=lfx_id_for(flow_id, version_no),
|
||||
replace=True,
|
||||
)
|
||||
except LfxInvalidFlow as exc:
|
||||
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
||||
|
|
|
|||
|
|
@ -1,15 +1,15 @@
|
|||
"""Authoring pipeline: Langflow export -> versioned definition -> LFX deploy."""
|
||||
"""Authoring pipeline: Langflow export -> canonical file -> version -> LFX deploy."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from sqlmodel import Session
|
||||
|
||||
from app.models import FlowDefinition, FlowVersion
|
||||
from app.settings import settings
|
||||
|
||||
from . import store
|
||||
from . import export, store, sync
|
||||
from .langflow_client import LangflowClient
|
||||
from .lfx_client import LfxClient
|
||||
from .schemas import FlowDefinitionCreate, FlowVersionCreate
|
||||
|
||||
|
||||
async def import_from_langflow(
|
||||
|
|
@ -22,39 +22,31 @@ async def import_from_langflow(
|
|||
author: str = "",
|
||||
deploy: bool = True,
|
||||
) -> tuple[FlowDefinition, FlowVersion]:
|
||||
"""Export a flow from Langflow, store it as a new version, deploy to LFX.
|
||||
"""Export a flow from Langflow, persist it to backend/flows, then sync.
|
||||
|
||||
Definitions are matched by name (explicit `name` wins, else the exported
|
||||
name); every import creates a new immutable version.
|
||||
The canonical file is written first so the directory stays the source of
|
||||
truth; sync_document() derives the definition + version from it.
|
||||
"""
|
||||
export = await langflow.export_flow(langflow_flow_id)
|
||||
flow_name = name or export.get("name") or f"langflow-{langflow_flow_id}"
|
||||
exported = await langflow.export_flow(langflow_flow_id)
|
||||
flow_name = name or exported.get("name") or f"langflow-{langflow_flow_id}"
|
||||
langflow_json = {
|
||||
"id": export.get("id", langflow_flow_id),
|
||||
"id": exported.get("id", langflow_flow_id),
|
||||
"name": flow_name,
|
||||
"description": export.get("description") or "",
|
||||
"endpoint_name": export.get("endpoint_name"),
|
||||
"data": export.get("data", {}),
|
||||
"tags": export.get("tags", []),
|
||||
"webhook": export.get("webhook", False),
|
||||
"is_component": export.get("is_component", False),
|
||||
"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),
|
||||
}
|
||||
try:
|
||||
definition = next(
|
||||
d for d in store.list_definitions(session) if d.name == flow_name
|
||||
)
|
||||
except StopIteration:
|
||||
definition = store.create_definition(
|
||||
session, FlowDefinitionCreate(name=flow_name)
|
||||
)
|
||||
version = store.create_version(
|
||||
path = export.write_flow_file(settings.flows_dir, langflow_json, name=flow_name)
|
||||
definition, version, _action = await sync.sync_document(
|
||||
session,
|
||||
definition.id,
|
||||
FlowVersionCreate(langflow_json=langflow_json, author=author),
|
||||
lfx,
|
||||
langflow_json,
|
||||
source_filename=path.name,
|
||||
author=author,
|
||||
deploy=deploy,
|
||||
)
|
||||
if deploy:
|
||||
uploaded = await lfx.upload_flow(
|
||||
version.langflow_json, flow_id=str(version.id), replace=True
|
||||
)
|
||||
version = store.mark_deployed(session, version, uploaded.id)
|
||||
assert version is not None # sync_document always versions on import
|
||||
return definition, version
|
||||
|
|
|
|||
71
backend/app/flows/export.py
Normal file
71
backend/app/flows/export.py
Normal file
|
|
@ -0,0 +1,71 @@
|
|||
"""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))
|
||||
|
|
@ -17,6 +17,8 @@ class FlowDefinitionRead(BaseModel):
|
|||
name: str
|
||||
description: str
|
||||
status: str
|
||||
# backend/flows filename this definition syncs from (None if API-created).
|
||||
source_path: str | None = None
|
||||
continuation_flow_id: UUID | None
|
||||
created_at: datetime
|
||||
|
||||
|
|
@ -110,3 +112,16 @@ class DecisionRequest(BaseModel):
|
|||
class DecisionRead(BaseModel):
|
||||
task: ApprovalRead
|
||||
continuation_run: RunRead | None
|
||||
|
||||
|
||||
class SyncReport(BaseModel):
|
||||
"""Result of reconciling backend/flows/*.json into DB + LFX.
|
||||
|
||||
Lists hold filename stems; errors maps filename 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)
|
||||
errors: dict[str, str] = Field(default_factory=dict)
|
||||
|
|
|
|||
|
|
@ -56,7 +56,82 @@ def list_definitions(session: Session) -> list[FlowDefinition]:
|
|||
return list(session.exec(select(FlowDefinition).order_by(FlowDefinition.name)).all())
|
||||
|
||||
|
||||
def _validate_flow_json(flow_json: dict) -> None:
|
||||
def get_definition_by_source_path(
|
||||
session: Session, source_path: str
|
||||
) -> FlowDefinition | None:
|
||||
"""Find the definition synced from a given backend/flows filename."""
|
||||
return session.exec(
|
||||
select(FlowDefinition).where(FlowDefinition.source_path == source_path)
|
||||
).first()
|
||||
|
||||
|
||||
def set_definition_name(
|
||||
session: Session, definition: FlowDefinition, name: str
|
||||
) -> FlowDefinition:
|
||||
"""Rename a definition and record its source file. Name clashes 409."""
|
||||
definition.name = name
|
||||
session.add(definition)
|
||||
try:
|
||||
session.commit()
|
||||
except IntegrityError as exc:
|
||||
session.rollback()
|
||||
raise ConflictError(f"flow {name!r} already exists") from exc
|
||||
session.refresh(definition)
|
||||
return definition
|
||||
|
||||
|
||||
def next_version_no(session: Session, flow_id: UUID) -> int:
|
||||
max_no = session.exec(
|
||||
select(func.max(FlowVersion.version_no)).where(FlowVersion.flow_id == flow_id)
|
||||
).one()
|
||||
return (max_no or 0) + 1
|
||||
|
||||
|
||||
def delete_definition_cascade(
|
||||
session: Session, definition: FlowDefinition
|
||||
) -> dict[str, int]:
|
||||
"""Remove a definition and its whole subtree (tasks, runs, versions).
|
||||
|
||||
Used when the canonical file disappears: the directory is the source of
|
||||
truth, so the derived registry follows. Commits once.
|
||||
"""
|
||||
version_ids = list(
|
||||
session.exec(
|
||||
select(FlowVersion.id).where(FlowVersion.flow_id == definition.id)
|
||||
).all()
|
||||
)
|
||||
task_count = 0
|
||||
run_count = 0
|
||||
if version_ids:
|
||||
tasks = list(
|
||||
session.exec(
|
||||
select(ApprovalTask).where(ApprovalTask.version_id.in_(version_ids))
|
||||
).all()
|
||||
)
|
||||
runs = list(
|
||||
session.exec(
|
||||
select(FlowRun).where(FlowRun.version_id.in_(version_ids))
|
||||
).all()
|
||||
)
|
||||
task_count, run_count = len(tasks), len(runs)
|
||||
for row in (*tasks, *runs):
|
||||
session.delete(row)
|
||||
for version in session.exec(
|
||||
select(FlowVersion).where(FlowVersion.flow_id == definition.id)
|
||||
).all():
|
||||
session.delete(version)
|
||||
session.delete(definition)
|
||||
session.commit()
|
||||
return {
|
||||
"definitions": 1,
|
||||
"versions": len(version_ids),
|
||||
"runs": run_count,
|
||||
"tasks": task_count,
|
||||
}
|
||||
|
||||
|
||||
def validate_flow_json(flow_json: dict) -> None:
|
||||
"""Reject anything that is not a Langflow-export-shaped document."""
|
||||
if not isinstance(flow_json, dict):
|
||||
raise InvalidError("langflow_json must be an object")
|
||||
data = flow_json.get("data")
|
||||
|
|
@ -70,7 +145,7 @@ def create_version(
|
|||
session: Session, flow_id: UUID, data: FlowVersionCreate
|
||||
) -> FlowVersion:
|
||||
get_definition(session, flow_id)
|
||||
_validate_flow_json(data.langflow_json)
|
||||
validate_flow_json(data.langflow_json)
|
||||
max_no = session.exec(
|
||||
select(func.max(FlowVersion.version_no)).where(FlowVersion.flow_id == flow_id)
|
||||
).one()
|
||||
|
|
|
|||
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),
|
||||
)
|
||||
|
|
@ -1,8 +1,58 @@
|
|||
import asyncio
|
||||
import logging
|
||||
from collections.abc import AsyncIterator
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import FastAPI
|
||||
|
||||
from .flows.api import approvals_router, flows_router
|
||||
from .flows.sync import run_sync_blocking, watch_forever
|
||||
from .settings import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _blocking_startup_sync() -> None:
|
||||
"""One-shot reconciliation at startup (fresh session, never fatal)."""
|
||||
try:
|
||||
report = run_sync_blocking(Path(settings.flows_dir))
|
||||
logger.info(
|
||||
"startup flow sync: +%d ~%d -%d skipped=%d errors=%d",
|
||||
len(report.added),
|
||||
len(report.updated),
|
||||
len(report.deleted),
|
||||
len(report.skipped),
|
||||
len(report.errors),
|
||||
)
|
||||
for filename, reason in report.errors.items():
|
||||
logger.warning("startup flow sync: %s: %s", filename, reason)
|
||||
except FileNotFoundError as exc:
|
||||
logger.error("startup flow sync skipped: %s", exc)
|
||||
except Exception: # noqa: BLE001 - API must start even if sync fails
|
||||
logger.exception("startup flow sync failed")
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
||||
task: asyncio.Task | None = None
|
||||
stop: asyncio.Event | None = None
|
||||
if settings.flows_watch_enabled:
|
||||
await asyncio.to_thread(_blocking_startup_sync)
|
||||
stop = asyncio.Event()
|
||||
task = asyncio.create_task(
|
||||
watch_forever(stop, Path(settings.flows_dir)), name="flows-watcher"
|
||||
)
|
||||
yield
|
||||
if task is not None and stop is not None:
|
||||
stop.set()
|
||||
task.cancel()
|
||||
try:
|
||||
await task
|
||||
except (asyncio.CancelledError, Exception): # noqa: BLE001 - shutdown
|
||||
pass
|
||||
|
||||
|
||||
app = FastAPI(
|
||||
title="Kirche Einfach API",
|
||||
description="KI-gestützte Verwaltungsplattform für Kirche",
|
||||
|
|
@ -10,6 +60,7 @@ app = FastAPI(
|
|||
docs_url="/api/docs",
|
||||
redoc_url="/api/redoc",
|
||||
openapi_url="/api/openapi.json",
|
||||
lifespan=lifespan,
|
||||
)
|
||||
|
||||
app.include_router(flows_router)
|
||||
|
|
|
|||
|
|
@ -29,6 +29,9 @@ class FlowDefinition(SQLModel, table=True):
|
|||
name: str = Field(index=True, unique=True, max_length=200)
|
||||
description: str = Field(default="", max_length=2000)
|
||||
status: str = Field(default="active", max_length=20) # active | archived
|
||||
# Filename under backend/flows/ this definition is synced from
|
||||
# (e.g. "simple_rule.json"). None for definitions created via the API.
|
||||
source_path: str | None = Field(default=None, max_length=255)
|
||||
# Flow to run after a human approval on this flow's runs (nullable).
|
||||
continuation_flow_id: UUID | None = Field(
|
||||
default=None, foreign_key="flow_definitions.id"
|
||||
|
|
|
|||
|
|
@ -13,10 +13,13 @@ class Settings(BaseSettings):
|
|||
version: str = "0.1.0"
|
||||
debug: bool = False
|
||||
|
||||
database_url: str = "postgresql+psycopg://kirche:kirche@localhost:5432/kirche_einfach"
|
||||
test_database_url: str = (
|
||||
"postgresql+psycopg://kirche:kirche@localhost:5432/kirche_einfach_test"
|
||||
)
|
||||
# App database (SQLite file). Override via DATABASE_URL for other engines.
|
||||
database_url: str = str(REPO_ROOT / "backend" / "data" / "kirche.db")
|
||||
|
||||
# Canonical flow definitions (source of truth): one Langflow export per file.
|
||||
flows_dir: str = str(REPO_ROOT / "backend" / "flows")
|
||||
# Auto-reconcile backend/flows into the DB + LFX while the API is running.
|
||||
flows_watch_enabled: bool = True
|
||||
|
||||
langflow_url: str = "http://localhost:7860"
|
||||
langflow_api_key: str = ""
|
||||
|
|
|
|||
1945
backend/flows/approval_handoff.json
Normal file
1945
backend/flows/approval_handoff.json
Normal file
File diff suppressed because one or more lines are too long
1393
backend/flows/simple_rule.json
Normal file
1393
backend/flows/simple_rule.json
Normal file
File diff suppressed because one or more lines are too long
|
|
@ -0,0 +1,32 @@
|
|||
"""add source_path to flow_definitions
|
||||
|
||||
Revision ID: d932b6993f9a
|
||||
Revises: 62ea9a5d8f9d
|
||||
Create Date: 2026-09-07 08:40:48.986026
|
||||
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
import sqlmodel # noqa: F401 - autogenerate renders sqlmodel.* column types
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = 'd932b6993f9a'
|
||||
down_revision: Union[str, Sequence[str], None] = '62ea9a5d8f9d'
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
"""Upgrade schema."""
|
||||
op.add_column(
|
||||
"flow_definitions",
|
||||
sa.Column("source_path", sa.String(length=255), nullable=True),
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""Downgrade schema."""
|
||||
op.drop_column("flow_definitions", "source_path")
|
||||
|
|
@ -7,10 +7,10 @@ dependencies = [
|
|||
"alembic>=1.14",
|
||||
"fastapi[standard]>=0.115.0",
|
||||
"httpx>=0.28",
|
||||
"psycopg[binary]>=3.2",
|
||||
"pydantic-settings>=2.0",
|
||||
"sqlmodel>=0.0.24",
|
||||
"uvicorn[standard]>=0.30.0",
|
||||
"watchfiles>=0.24",
|
||||
]
|
||||
|
||||
[build-system]
|
||||
|
|
@ -27,7 +27,7 @@ testpaths = ["tests"]
|
|||
|
||||
[dependency-groups]
|
||||
dev = [
|
||||
"lfx==1.12.0",
|
||||
"lfx",
|
||||
"pytest>=8",
|
||||
"pytest-asyncio>=0.24",
|
||||
"pytest-cov>=6",
|
||||
|
|
|
|||
|
|
@ -6,6 +6,10 @@ Langflow UI would export. Run with the backend venv (has the `lfx` dev dep)::
|
|||
|
||||
uv run python scripts/make_flow_fixtures.py
|
||||
|
||||
Writes to two places:
|
||||
- backend/tests/fixtures/flows/ (test fixtures, never deployed as-is)
|
||||
- backend/flows/ (canonical source of truth, synced into DB + LFX)
|
||||
|
||||
Then verify::
|
||||
|
||||
uv run lfx validate --level 4 backend/tests/fixtures/flows/
|
||||
|
|
@ -21,6 +25,7 @@ from pathlib import Path
|
|||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent.parent
|
||||
FIXTURE_DIR = REPO_ROOT / "backend" / "tests" / "fixtures" / "flows"
|
||||
FLOW_DIR = REPO_ROOT / "backend" / "flows"
|
||||
|
||||
HANDLE_QUOTE = "\u0153" # ReactFlow handle-id quoting used by Langflow exports
|
||||
|
||||
|
|
@ -430,14 +435,16 @@ def build_approval_handoff() -> dict:
|
|||
|
||||
def main() -> None:
|
||||
FIXTURE_DIR.mkdir(parents=True, exist_ok=True)
|
||||
FLOW_DIR.mkdir(parents=True, exist_ok=True)
|
||||
flows = {
|
||||
"simple_rule.json": build_simple_rule(),
|
||||
"approval_handoff.json": build_approval_handoff(),
|
||||
}
|
||||
for filename, doc in flows.items():
|
||||
path = FIXTURE_DIR / filename
|
||||
path.write_text(json.dumps(doc, indent=2, ensure_ascii=False) + "\n")
|
||||
print(f"wrote {path}")
|
||||
payload = json.dumps(doc, indent=2, ensure_ascii=False) + "\n"
|
||||
for target in (FIXTURE_DIR / filename, FLOW_DIR / filename):
|
||||
target.write_text(payload)
|
||||
print(f"wrote {target}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
|
|
@ -1,7 +1,8 @@
|
|||
"""Manual end-to-end smoke: real LFX + real API, frontend-shaped calls.
|
||||
|
||||
Usage (from backend/):
|
||||
TEST_DATABASE_URL=postgresql+psycopg://... uv run python scripts/smoke_e2e.py
|
||||
uv run python scripts/smoke_e2e.py
|
||||
Runs the API against a throwaway SQLite file and an ephemeral LFX pass.
|
||||
Exits non-zero on any unexpected status/body.
|
||||
"""
|
||||
|
||||
|
|
@ -11,7 +12,7 @@ import json
|
|||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import tempfile
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
|
|
@ -54,17 +55,31 @@ def check(name: str, cond: bool, detail: str = "") -> None:
|
|||
|
||||
|
||||
def main() -> None:
|
||||
tmp = Path(tempfile.mkdtemp(prefix="kirche-smoke-"))
|
||||
flow_dir = Path("/tmp/smoke-e2e-flows")
|
||||
flow_dir.mkdir(exist_ok=True)
|
||||
(flow_dir / "seed.json").write_text(
|
||||
(FIXTURES / "simple_rule.json").read_text()
|
||||
)
|
||||
# Migrate a throwaway SQLite schema to head before starting the API.
|
||||
from alembic import command as alembic_command
|
||||
from alembic.config import Config
|
||||
|
||||
db_url = f"sqlite:///{tmp / 'smoke.db'}"
|
||||
cfg = Config()
|
||||
cfg.set_main_option("script_location", str(BACKEND / "migrations"))
|
||||
cfg.set_main_option("sqlalchemy.url", db_url)
|
||||
alembic_command.upgrade(cfg, "head")
|
||||
|
||||
env = {
|
||||
**os.environ,
|
||||
# lfx prefers LANGFLOW_API_KEY over LFX_API_KEY; the repo .env may set
|
||||
# it. Clear it so the serve key below is the one we send.
|
||||
"LANGFLOW_API_KEY": "",
|
||||
"LFX_API_KEY": LFX_KEY,
|
||||
"DATABASE_URL": os.environ["TEST_DATABASE_URL"],
|
||||
"DATABASE_URL": db_url,
|
||||
"FLOWS_WATCH_ENABLED": "false",
|
||||
"LFX_URL": f"http://127.0.0.1:{LFX_PORT}",
|
||||
"LFX_API_KEY": LFX_KEY,
|
||||
"CALLBACK_BASE_URL": f"http://127.0.0.1:{API_PORT}",
|
||||
}
|
||||
lfx = subprocess.Popen(
|
||||
|
|
|
|||
89
backend/scripts/sync_flows.py
Normal file
89
backend/scripts/sync_flows.py
Normal file
|
|
@ -0,0 +1,89 @@
|
|||
"""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
|
||||
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 Exception as exc: # noqa: BLE001 - pull-all is total
|
||||
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())
|
||||
|
|
@ -1,25 +1,25 @@
|
|||
"""Shared pytest fixtures.
|
||||
|
||||
Requires TEST_DATABASE_URL (a dedicated Postgres database, e.g. the
|
||||
`kirche_einfach_test` database from the compose stack). The URL is mapped
|
||||
onto DATABASE_URL *before* any app module is imported, so settings and the
|
||||
alembic env both target the test database. Never point this at production.
|
||||
Tests run on a throwaway SQLite file (no external database required).
|
||||
`DATABASE_URL` and `FLOWS_DIR` are pointed at a session temp dir *before* any
|
||||
app module is imported, so settings pick them up at import time. The app
|
||||
schema is migrated to head once per session via Alembic.
|
||||
"""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
TEST_DB_URL = os.environ.get("TEST_DATABASE_URL")
|
||||
if not TEST_DB_URL:
|
||||
raise RuntimeError(
|
||||
"TEST_DATABASE_URL is not set. Refusing to run tests without "
|
||||
"an explicit test database (see .env.example)."
|
||||
)
|
||||
TEST_AREA = Path(tempfile.mkdtemp(prefix="kirche-tests-"))
|
||||
|
||||
# Must precede all `app.*` imports: Settings is instantiated at import time.
|
||||
os.environ["DATABASE_URL"] = TEST_DB_URL
|
||||
os.environ["DATABASE_URL"] = f"sqlite:///{TEST_AREA / 'test.db'}"
|
||||
os.environ["FLOWS_DIR"] = str(TEST_AREA / "flows")
|
||||
# The embedded watcher + startup sync stay off in tests; sync is exercised
|
||||
# directly (sync_from_directory) with per-test tmp dirs instead.
|
||||
os.environ["FLOWS_WATCH_ENABLED"] = "false"
|
||||
|
||||
BACKEND_DIR = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
|
@ -115,6 +115,10 @@ def lfx_server(tmp_path_factory):
|
|||
env = {
|
||||
**os.environ,
|
||||
# lfx serve refuses to start without a serve key; requests need it too.
|
||||
# get_api_key() prefers LANGFLOW_API_KEY over LFX_API_KEY, and the repo
|
||||
# .env may put a real Langflow key in the environment — clear it so the
|
||||
# test key below is the one the client sends.
|
||||
"LANGFLOW_API_KEY": "",
|
||||
"LFX_API_KEY": "test-lfx-key",
|
||||
}
|
||||
proc = subprocess.Popen(
|
||||
|
|
|
|||
|
|
@ -2,7 +2,9 @@
|
|||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
from app.flows.sync import lfx_id_for
|
||||
|
||||
FIXTURES = Path(__file__).parent / "fixtures" / "flows"
|
||||
LANGFLOW_ID = "11111111-2222-4333-8444-555555555555"
|
||||
|
|
@ -35,7 +37,9 @@ async def test_import_creates_definition_version_and_deploys(
|
|||
assert body["definition"]["name"] == "importiert"
|
||||
assert body["version"]["version_no"] == 1
|
||||
assert body["version"]["author"] == "admin"
|
||||
assert body["version"]["lfx_flow_id"] == body["version"]["id"]
|
||||
assert body["version"]["lfx_flow_id"] == lfx_id_for(
|
||||
UUID(body["definition"]["id"]), 1
|
||||
)
|
||||
assert body["version"]["deployed_at"] is not None
|
||||
assert body["version"]["langflow_json"]["data"]["nodes"]
|
||||
|
||||
|
|
@ -48,7 +52,7 @@ async def test_import_creates_definition_version_and_deploys(
|
|||
assert response.json()["status"] == "completed"
|
||||
|
||||
|
||||
async def test_import_reuses_definition_and_bumps_version(
|
||||
async def test_import_same_content_is_idempotent(
|
||||
client, lfx_server, override_langflow
|
||||
):
|
||||
override_langflow(_export_doc())
|
||||
|
|
@ -61,9 +65,44 @@ async def test_import_reuses_definition_and_bumps_version(
|
|||
assert first.status_code == 201
|
||||
assert second.status_code == 201
|
||||
assert first.json()["definition"]["id"] == second.json()["definition"]["id"]
|
||||
# Unchanged content creates no new version.
|
||||
assert second.json()["version"]["version_no"] == 1
|
||||
|
||||
|
||||
async def test_import_changed_content_bumps_version(
|
||||
client, lfx_server, override_langflow
|
||||
):
|
||||
override_langflow(_export_doc())
|
||||
first = await client.post(
|
||||
"/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID}
|
||||
)
|
||||
assert first.status_code == 201
|
||||
changed = _export_doc()
|
||||
changed["description"] = "neue Beschreibung"
|
||||
override_langflow(changed)
|
||||
second = await client.post(
|
||||
"/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID}
|
||||
)
|
||||
assert second.status_code == 201
|
||||
assert second.json()["version"]["version_no"] == 2
|
||||
|
||||
|
||||
async def test_import_writes_canonical_file(
|
||||
client, lfx_server, override_langflow, tmp_path, monkeypatch
|
||||
):
|
||||
from app.settings import settings
|
||||
|
||||
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
||||
override_langflow(_export_doc())
|
||||
response = await client.post(
|
||||
"/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID}
|
||||
)
|
||||
assert response.status_code == 201, response.text
|
||||
written = tmp_path / "importiert.json"
|
||||
assert written.is_file()
|
||||
assert json.loads(written.read_text())["name"] == "importiert"
|
||||
|
||||
|
||||
async def test_import_with_explicit_name(client, lfx_server, override_langflow):
|
||||
override_langflow(_export_doc(name="langflow-name"))
|
||||
response = await client.post(
|
||||
|
|
|
|||
39
backend/tests/test_export.py
Normal file
39
backend/tests/test_export.py
Normal file
|
|
@ -0,0 +1,39 @@
|
|||
"""Unit tests for canonical flow files (export.py). No DB or LFX needed."""
|
||||
|
||||
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_canonical_bytes_stable_under_key_reorder():
|
||||
doc = {"name": "x", "data": {"nodes": [], "edges": []}, "tags": ["b", "a"]}
|
||||
shuffled = {"tags": ["b", "a"], "data": {"edges": [], "nodes": []}, "name": "x"}
|
||||
assert export.canonical_bytes(shuffled) == export.canonical_bytes(doc)
|
||||
assert export.content_hash(doc) == export.content_hash(shuffled)
|
||||
|
||||
|
||||
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 json.loads(path.read_text()) == doc
|
||||
# Rewriting identical content is byte-identical (no phantom git diffs).
|
||||
before = path.read_bytes()
|
||||
export.write_flow_file(tmp_path, doc)
|
||||
assert path.read_bytes() == before
|
||||
|
||||
|
||||
def test_iter_flow_files_only_top_level_json(tmp_path):
|
||||
(tmp_path / "a.json").write_text("{}")
|
||||
(tmp_path / "b.txt").write_text("{}")
|
||||
sub = tmp_path / "sub"
|
||||
sub.mkdir()
|
||||
(sub / "c.json").write_text("{}")
|
||||
assert export.iter_flow_files(tmp_path) == [tmp_path / "a.json"]
|
||||
|
|
@ -2,7 +2,9 @@
|
|||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
from app.flows.sync import lfx_id_for
|
||||
|
||||
FIXTURES = Path(__file__).parent / "fixtures" / "flows"
|
||||
|
||||
|
|
@ -81,7 +83,7 @@ async def test_deploy_and_run_standard_and_sonderfall(client, lfx_server):
|
|||
flow = await _create_flow(client, "dachsanierung")
|
||||
await _create_version(client, flow["id"])
|
||||
deployed = await _deploy(client, flow["id"])
|
||||
assert deployed["lfx_flow_id"] == deployed["id"]
|
||||
assert deployed["lfx_flow_id"] == lfx_id_for(UUID(flow["id"]), 1)
|
||||
assert deployed["deployed_at"] is not None
|
||||
|
||||
response = await client.post(
|
||||
|
|
|
|||
|
|
@ -16,7 +16,9 @@ def test_flow_definition_defaults(session):
|
|||
assert d.status == "active"
|
||||
assert d.continuation_flow_id is None
|
||||
assert d.created_at is not None
|
||||
assert d.created_at.tzinfo is not None
|
||||
# Timestamps are written as UTC. SQLite does not round-trip tzinfo, so
|
||||
# after a refresh the value is calendar-time UTC without tzinfo.
|
||||
assert d.created_at.year >= 2026
|
||||
|
||||
|
||||
def test_flow_definition_name_unique(session):
|
||||
|
|
|
|||
157
backend/tests/test_sync.py
Normal file
157
backend/tests/test_sync.py
Normal file
|
|
@ -0,0 +1,157 @@
|
|||
"""Sync engine tests: backend/flows/*.json -> DB + LFX (LFX real, DB sqlite)."""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import shutil
|
||||
import time
|
||||
from pathlib import Path
|
||||
from uuid import UUID
|
||||
|
||||
import pytest
|
||||
|
||||
from app.flows import store, sync
|
||||
from app.flows.lfx_client import LfxClient, LfxNotFound
|
||||
from app.flows.schemas import SyncReport
|
||||
from app.flows.sync import lfx_id_for
|
||||
from app.settings import settings
|
||||
|
||||
FIXTURES = Path(__file__).parent / "fixtures" / "flows"
|
||||
|
||||
|
||||
def load_fixture(name: str) -> dict:
|
||||
return json.loads((FIXTURES / name).read_text())
|
||||
|
||||
|
||||
def _lfx() -> LfxClient:
|
||||
return LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None)
|
||||
|
||||
|
||||
def _seed_dir(tmp_path: Path, name: str = "simple_rule.json") -> Path:
|
||||
shutil.copy(FIXTURES / name, tmp_path / name)
|
||||
return tmp_path
|
||||
|
||||
|
||||
async def test_sync_adds_new_flow(session, lfx_server, tmp_path):
|
||||
_seed_dir(tmp_path)
|
||||
report = await sync.sync_from_directory(session, _lfx(), tmp_path)
|
||||
assert report.added == ["simple_rule"]
|
||||
assert report.errors == {}
|
||||
definition = store.get_definition_by_source_path(session, "simple_rule.json")
|
||||
assert definition is not None
|
||||
assert definition.name == "simple_rule"
|
||||
versions = store.list_versions(session, definition.id)
|
||||
assert [v.version_no for v in versions] == [1]
|
||||
assert versions[0].lfx_flow_id == lfx_id_for(definition.id, 1)
|
||||
assert versions[0].deployed_at is not None
|
||||
|
||||
|
||||
async def test_sync_skips_unchanged(session, lfx_server, tmp_path):
|
||||
_seed_dir(tmp_path)
|
||||
first = await sync.sync_from_directory(session, _lfx(), tmp_path)
|
||||
assert first.added == ["simple_rule"]
|
||||
second = await sync.sync_from_directory(session, _lfx(), tmp_path)
|
||||
assert second.skipped == ["simple_rule"]
|
||||
assert second.added == second.updated == []
|
||||
definition = store.get_definition_by_source_path(session, "simple_rule.json")
|
||||
assert len(store.list_versions(session, definition.id)) == 1
|
||||
|
||||
|
||||
async def test_sync_updates_on_change_and_replaces_lfx(session, lfx_server, tmp_path):
|
||||
_seed_dir(tmp_path)
|
||||
await sync.sync_from_directory(session, _lfx(), tmp_path)
|
||||
definition = store.get_definition_by_source_path(session, "simple_rule.json")
|
||||
old_lfx_id = lfx_id_for(definition.id, 1)
|
||||
|
||||
doc = load_fixture("simple_rule.json")
|
||||
doc["description"] = "geaendert"
|
||||
(tmp_path / "simple_rule.json").write_text(json.dumps(doc))
|
||||
|
||||
report = await sync.sync_from_directory(session, _lfx(), tmp_path)
|
||||
assert report.updated == ["simple_rule"]
|
||||
versions = store.list_versions(session, definition.id)
|
||||
assert [v.version_no for v in versions] == [1, 2]
|
||||
assert versions[1].lfx_flow_id == lfx_id_for(definition.id, 2)
|
||||
# Superseded LFX registration is gone.
|
||||
with pytest.raises(LfxNotFound):
|
||||
await _lfx().flow_info(old_lfx_id)
|
||||
|
||||
|
||||
async def test_sync_deletes_missing_file_with_cascade(session, lfx_server, tmp_path):
|
||||
_seed_dir(tmp_path)
|
||||
await sync.sync_from_directory(session, _lfx(), tmp_path)
|
||||
definition = store.get_definition_by_source_path(session, "simple_rule.json")
|
||||
version = store.list_versions(session, definition.id)[0]
|
||||
run = store.create_run(session, version, input={"a": 1}, tenant_id="default")
|
||||
lfx_id = lfx_id_for(definition.id, 1)
|
||||
|
||||
(tmp_path / "simple_rule.json").unlink()
|
||||
report = await sync.sync_from_directory(session, _lfx(), tmp_path)
|
||||
assert report.deleted == ["simple_rule"]
|
||||
assert store.get_definition_by_source_path(session, "simple_rule.json") is None
|
||||
with pytest.raises(store.NotFoundError):
|
||||
store.get_run(session, run.id)
|
||||
with pytest.raises(LfxNotFound):
|
||||
await _lfx().flow_info(lfx_id)
|
||||
|
||||
|
||||
async def test_sync_reports_invalid_json(session, lfx_server, tmp_path):
|
||||
(tmp_path / "kaputt.json").write_text("{kein json")
|
||||
report = await sync.sync_from_directory(session, _lfx(), tmp_path)
|
||||
assert "kaputt.json" in report.errors
|
||||
assert report.added == []
|
||||
assert store.list_definitions(session) == []
|
||||
|
||||
|
||||
async def test_sync_missing_dir_raises(session, lfx_server, tmp_path):
|
||||
with pytest.raises(FileNotFoundError):
|
||||
await sync.sync_from_directory(session, _lfx(), tmp_path / "gibts-nicht")
|
||||
|
||||
|
||||
async def test_sync_rename_updates_definition_name(session, lfx_server, tmp_path):
|
||||
_seed_dir(tmp_path)
|
||||
await sync.sync_from_directory(session, _lfx(), tmp_path)
|
||||
doc = load_fixture("simple_rule.json")
|
||||
doc["name"] = "umbenannt"
|
||||
(tmp_path / "simple_rule.json").write_text(json.dumps(doc))
|
||||
report = await sync.sync_from_directory(session, _lfx(), tmp_path)
|
||||
assert report.updated == ["simple_rule"]
|
||||
definition = store.get_definition_by_source_path(session, "simple_rule.json")
|
||||
assert definition.name == "umbenannt"
|
||||
|
||||
|
||||
async def test_sync_endpoint_returns_report(client, lfx_server, tmp_path, monkeypatch):
|
||||
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["errors"] == {}
|
||||
|
||||
response = await client.post("/api/flows/sync")
|
||||
assert response.json()["skipped"] == ["simple_rule"]
|
||||
|
||||
|
||||
async def test_watch_forever_triggers_on_file_change(tmp_path):
|
||||
import contextlib
|
||||
|
||||
calls: list[str] = []
|
||||
stop = asyncio.Event()
|
||||
|
||||
def run():
|
||||
calls.append("sync")
|
||||
return SyncReport()
|
||||
|
||||
task = asyncio.create_task(sync.watch_forever(stop, tmp_path, run_sync=run))
|
||||
try:
|
||||
await asyncio.sleep(0.5) # let the watcher arm
|
||||
(tmp_path / "neu.json").write_text("{}")
|
||||
deadline = time.time() + 15
|
||||
while not calls and time.time() < deadline:
|
||||
await asyncio.sleep(0.1)
|
||||
assert calls == ["sync"]
|
||||
finally:
|
||||
stop.set()
|
||||
task.cancel()
|
||||
with contextlib.suppress(asyncio.CancelledError):
|
||||
await task
|
||||
122
compose.prod.yml
Normal file
122
compose.prod.yml
Normal file
|
|
@ -0,0 +1,122 @@
|
|||
# Minimal self-hosted deployment (internal use: no TLS / reverse proxy).
|
||||
#
|
||||
# Flows are the bind-mounted ./backend/flows directory (source of truth);
|
||||
# the embedded API watcher reconciles them into the DB + LFX automatically.
|
||||
# Registry images (:latest) auto-update nightly via watchtower; locally
|
||||
# built images (app, lfx, frontend) update via `docker compose build --pull`.
|
||||
#
|
||||
# Usage: docker compose -f compose.prod.yml up -d --build
|
||||
services:
|
||||
app:
|
||||
build:
|
||||
context: ./backend
|
||||
dockerfile: Dockerfile
|
||||
env_file:
|
||||
- .env
|
||||
environment:
|
||||
DATABASE_URL: sqlite:////srv/backend/data/kirche.db
|
||||
FLOWS_DIR: /srv/backend/flows
|
||||
LFX_URL: http://lfx:6004
|
||||
LANGFLOW_URL: http://langflow:7860
|
||||
CALLBACK_BASE_URL: http://app:8000
|
||||
volumes:
|
||||
- ./backend/flows:/srv/backend/flows
|
||||
- appdata:/srv/backend/data
|
||||
ports:
|
||||
- "8000:8000"
|
||||
depends_on:
|
||||
lfx:
|
||||
condition: service_healthy
|
||||
langflow:
|
||||
condition: service_healthy
|
||||
restart: unless-stopped
|
||||
healthcheck:
|
||||
test:
|
||||
[
|
||||
"CMD-SHELL",
|
||||
"python -c \"import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=5)\"",
|
||||
]
|
||||
interval: 30s
|
||||
timeout: 5s
|
||||
retries: 3
|
||||
start_period: 30s
|
||||
|
||||
frontend:
|
||||
build:
|
||||
context: ./frontend
|
||||
dockerfile: Dockerfile
|
||||
ports:
|
||||
- "3000:3000"
|
||||
depends_on:
|
||||
app:
|
||||
condition: service_healthy
|
||||
restart: unless-stopped
|
||||
|
||||
langflow:
|
||||
image: langflowai/langflow:latest
|
||||
env_file:
|
||||
- .env
|
||||
environment:
|
||||
LANGFLOW_CONFIG_DIR: /app/langflow
|
||||
LANGFLOW_HOST: 0.0.0.0
|
||||
PORT: "7860"
|
||||
LANGFLOW_AUTO_LOGIN: "false"
|
||||
LANGFLOW_API_KEY_SOURCE: env
|
||||
LANGFLOW_DATABASE_URL: sqlite:////app/langflow/langflow.db
|
||||
volumes:
|
||||
- langflow-data:/app/langflow
|
||||
ports:
|
||||
- "7860:7860"
|
||||
labels:
|
||||
com.centurylinklabs.watchtower.enable: "true"
|
||||
restart: unless-stopped
|
||||
healthcheck:
|
||||
test:
|
||||
[
|
||||
"CMD-SHELL",
|
||||
"python -c \"import urllib.request; urllib.request.urlopen('http://localhost:7860/health', timeout=5)\"",
|
||||
]
|
||||
interval: 30s
|
||||
timeout: 5s
|
||||
retries: 20
|
||||
start_period: 60s
|
||||
|
||||
lfx:
|
||||
build:
|
||||
context: .devcontainer
|
||||
dockerfile: Dockerfile.lfx
|
||||
env_file:
|
||||
- .env
|
||||
volumes:
|
||||
- lfx-data:/flows
|
||||
ports:
|
||||
- "6004:6004"
|
||||
restart: unless-stopped
|
||||
healthcheck:
|
||||
test:
|
||||
[
|
||||
"CMD-SHELL",
|
||||
"python -c \"import urllib.request; urllib.request.urlopen('http://localhost:6004/health', timeout=5)\"",
|
||||
]
|
||||
interval: 30s
|
||||
timeout: 5s
|
||||
retries: 12
|
||||
start_period: 30s
|
||||
|
||||
watchtower:
|
||||
image: nickfedor/watchtower:latest
|
||||
volumes:
|
||||
- /var/run/docker.sock:/var/run/docker.sock
|
||||
environment:
|
||||
WATCHTOWER_LABEL_ENABLE: "true"
|
||||
WATCHTOWER_SCHEDULE: "0 0 4 * * *"
|
||||
WATCHTOWER_CLEANUP: "true"
|
||||
WATCHTOWER_ROLLING_RESTART: "true"
|
||||
labels:
|
||||
com.centurylinklabs.watchtower.enable: "true"
|
||||
restart: unless-stopped
|
||||
|
||||
volumes:
|
||||
appdata:
|
||||
langflow-data:
|
||||
lfx-data:
|
||||
13
frontend/Dockerfile
Normal file
13
frontend/Dockerfile
Normal file
|
|
@ -0,0 +1,13 @@
|
|||
# Production image for the React UI (build + Vite preview; internal use).
|
||||
FROM node:22-slim
|
||||
|
||||
WORKDIR /srv/frontend
|
||||
COPY package.json package-lock.json ./
|
||||
RUN npm ci
|
||||
COPY . ./
|
||||
RUN npm run build
|
||||
|
||||
ENV VITE_API_PROXY_TARGET=http://app:8000
|
||||
EXPOSE 3000
|
||||
|
||||
CMD ["npm", "run", "preview", "--", "--host", "0.0.0.0", "--port", "3000"]
|
||||
|
|
@ -5,6 +5,7 @@ export interface FlowDefinition {
|
|||
name: string;
|
||||
description: string;
|
||||
status: string;
|
||||
source_path: string | null;
|
||||
continuation_flow_id: string | null;
|
||||
created_at: string;
|
||||
}
|
||||
|
|
@ -60,6 +61,14 @@ export interface Decision {
|
|||
continuation_run: Run | null;
|
||||
}
|
||||
|
||||
export interface SyncReport {
|
||||
added: string[];
|
||||
updated: string[];
|
||||
deleted: string[];
|
||||
skipped: string[];
|
||||
errors: Record<string, string>;
|
||||
}
|
||||
|
||||
export class ApiError extends Error {
|
||||
status: number;
|
||||
constructor(status: number, message: string) {
|
||||
|
|
@ -135,4 +144,8 @@ export const api = {
|
|||
),
|
||||
listLangflow: () =>
|
||||
req<{ id: string; name: string }[]>("/api/flows/langflow/available"),
|
||||
syncFlows: () =>
|
||||
req<SyncReport>("/api/flows/sync", {
|
||||
method: "POST",
|
||||
}),
|
||||
};
|
||||
|
|
|
|||
|
|
@ -64,6 +64,26 @@ export default function AdminTab() {
|
|||
}
|
||||
}
|
||||
|
||||
async function onSync() {
|
||||
setError("");
|
||||
setInfo("");
|
||||
try {
|
||||
const r = await api.syncFlows();
|
||||
const parts = [
|
||||
`+${r.added.length}`,
|
||||
`~${r.updated.length}`,
|
||||
`-${r.deleted.length}`,
|
||||
].join(" ");
|
||||
const errs = Object.entries(r.errors)
|
||||
.map(([f, e]) => `${f}: ${e}`)
|
||||
.join("; ");
|
||||
setInfo(`Sync ${parts}${errs ? ` — Fehler: ${errs}` : ""}`);
|
||||
await load();
|
||||
} catch (e) {
|
||||
setError(e instanceof ApiError ? `${e.status}: ${e.message}` : String(e));
|
||||
}
|
||||
}
|
||||
|
||||
return (
|
||||
<div>
|
||||
<h2>Administration</h2>
|
||||
|
|
@ -93,11 +113,19 @@ export default function AdminTab() {
|
|||
</button>
|
||||
</div>
|
||||
<h3>Flows + Versionen</h3>
|
||||
<div style={{ marginBottom: 12 }}>
|
||||
<button onClick={() => void onSync()}>
|
||||
Aus backend/flows synchronisieren
|
||||
</button>
|
||||
</div>
|
||||
{flows.length === 0 && <p>Keine Flows vorhanden.</p>}
|
||||
<ul>
|
||||
{flows.map((f) => (
|
||||
<li key={f.id} style={{ marginBottom: 8 }}>
|
||||
<strong>{f.name}</strong>
|
||||
<strong>{f.name}</strong>{" "}
|
||||
<span style={{ fontFamily: "monospace", fontSize: "0.85em" }}>
|
||||
{f.source_path ? `(${f.source_path})` : "(API)"}
|
||||
</span>
|
||||
<ul>
|
||||
{(versions[f.id] ?? []).map((v) => (
|
||||
<li key={v.id}>
|
||||
|
|
|
|||
|
|
@ -12,4 +12,14 @@ export default defineConfig({
|
|||
},
|
||||
},
|
||||
},
|
||||
preview: {
|
||||
port: 3000,
|
||||
proxy: {
|
||||
"/api": {
|
||||
// Overridable at container start (prod compose: http://app:8000).
|
||||
target: process.env.VITE_API_PROXY_TARGET ?? "http://localhost:8000",
|
||||
changeOrigin: true,
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue