diff --git a/CHANGELOG.md b/CHANGELOG.md index a0356b0..b76708c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,11 @@ 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] +### Changed +- Declutter: shared `BaseApiClient` (LFX/Langflow), single `get_db`, `model_validate` mappers, unified LFX/Langflow error mapping, stream logic moved to `runner.py`, `sync_full_pass` single orchestration (API+CLI), shared model base (`utcnow`/`ts_column`), unified auth counters, `httpx` in `deploy_flow.py`. +- Forms: shared `components/kirche/_form_core.py` single source (codegen-inlined for LFX), structured `visible_if` key alongside legacy markers, `LangflowClient.upsert_flow` + `scripts/push_flows_to_langflow.py` (6 canonical flows pushed to Langflow UI). +- Tests (full real LFX kept, 146 green): shared `conftest` helpers (`load_flow_fixture`, `flow_helpers`, `lfx_client`, `anon_client`), `tests/_harness.py` shared with smoke script, parametrized `filename_stem`. +- Frontend: shared `formatApiError`, ``, ``, `URLSearchParams` queries, localized run dates. ### Added - Branched questionnaires: `[visible_if feld=Wert]` markers in a `CustomInputForm` field's `info` (code editor) show the field only when diff --git a/backend/app/auth/api.py b/backend/app/auth/api.py index 7880082..5c20ff6 100644 --- a/backend/app/auth/api.py +++ b/backend/app/auth/api.py @@ -6,16 +6,11 @@ from uuid import UUID from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response from sqlmodel import Session +from app.db import get_db from app.models import User from . import mail, service -from .deps import ( - SESSION_COOKIE, - SESSION_MAX_AGE, - get_current_user, - get_db, - require_admin, -) +from .deps import SESSION_COOKIE, SESSION_MAX_AGE, get_current_user, require_admin from .schemas import ( AccessDecisionRead, AccessRequestCreate, @@ -33,26 +28,11 @@ auth_router = APIRouter(prefix="/api/auth", tags=["auth"]) def _read_request(r) -> AccessRequestRead: - return AccessRequestRead( - id=r.id, - name=r.name, - email=r.email, - reason=r.reason, - status=r.status, - created_at=r.created_at, - decided_at=r.decided_at, - decided_by=r.decided_by, - ) + return AccessRequestRead.model_validate(r, from_attributes=True) def _read_me(user: User) -> MeRead: - return MeRead( - id=user.id, - email=user.email, - name=user.name, - role=user.role, - tenant_id=user.tenant_id, - ) + return MeRead.model_validate(user, from_attributes=True) def _set_session_cookie(response: Response, session_id: str) -> None: diff --git a/backend/app/auth/deps.py b/backend/app/auth/deps.py index b090eb9..2fc921c 100644 --- a/backend/app/auth/deps.py +++ b/backend/app/auth/deps.py @@ -1,14 +1,13 @@ -"""Session-cookie dependencies. Handlers stay thin; DB via get_session.""" +"""Session-cookie dependencies. Handlers stay thin; DB via app.db.get_db.""" from __future__ import annotations -from collections.abc import Iterator from typing import Annotated from fastapi import Depends, HTTPException, Request from sqlmodel import Session -from app.db import get_session +from app.db import get_db from app.models import User from . import service @@ -17,10 +16,6 @@ SESSION_COOKIE = "kirche_session" SESSION_MAX_AGE = int(service.SESSION_TTL.total_seconds()) -def get_db() -> Iterator[Session]: - yield from get_session() - - def get_optional_user( request: Request, session: Annotated[Session, Depends(get_db)] ) -> User | None: diff --git a/backend/app/auth/mail.py b/backend/app/auth/mail.py index a8cf957..b50e3a8 100644 --- a/backend/app/auth/mail.py +++ b/backend/app/auth/mail.py @@ -22,11 +22,6 @@ BLUE_LIGHT = "#eef4fb" INK = "#162f4c" MUTED = "#5b6b7c" -BLUE = "#1d4a79" -BLUE_LIGHT = "#eef4fb" -INK = "#162f4c" -MUTED = "#5b6b7c" - def magic_link(token: str) -> str: return f"{settings.frontend_url.rstrip('/')}/login?token={token}" diff --git a/backend/app/auth/service.py b/backend/app/auth/service.py index 8681284..3a501d1 100644 --- a/backend/app/auth/service.py +++ b/backend/app/auth/service.py @@ -15,6 +15,7 @@ from uuid import UUID from sqlmodel import Session, func, select from app.models import AccessRequest, MagicLinkToken, User, UserSession +from app.models.base import utcnow from app.settings import settings logger = logging.getLogger(__name__) @@ -60,10 +61,6 @@ class RateLimitedError(AuthError): pass -def utcnow() -> datetime: - return datetime.now(UTC) - - def normalize_email(email: str) -> str: return email.strip().lower() @@ -92,18 +89,22 @@ def user_count(session: Session) -> int: return session.exec(select(func.count()).select_from(User)).one() -def recent_request_count(session: Session, email: str, hours: int = 1) -> int: +def _recent_count(session: Session, model, *filters, hours: int = 1) -> int: since = utcnow() - timedelta(hours=hours) return session.exec( - select(func.count()) - .select_from(AccessRequest) - .where( - AccessRequest.email == normalize_email(email), - AccessRequest.created_at >= since, - ) + select(func.count()).select_from(model).where(*filters, model.created_at >= since) ).one() +def recent_request_count(session: Session, email: str, hours: int = 1) -> int: + return _recent_count( + session, + AccessRequest, + AccessRequest.email == normalize_email(email), + hours=hours, + ) + + def create_access_request( session: Session, name: str, email: str, reason: str = "" ) -> AccessRequest: @@ -165,15 +166,7 @@ def issue_magic_token(session: Session, user: User) -> str: def recent_token_count(session: Session, user: User, hours: int = 1) -> int: - since = utcnow() - timedelta(hours=hours) - return session.exec( - select(func.count()) - .select_from(MagicLinkToken) - .where( - MagicLinkToken.user_id == user.id, - MagicLinkToken.created_at >= since, - ) - ).one() + return _recent_count(session, MagicLinkToken, MagicLinkToken.user_id == user.id, hours=hours) def request_magic_link(session: Session, email: str) -> tuple[User, str]: diff --git a/backend/app/db.py b/backend/app/db.py index 768c2ea..c3aa192 100644 --- a/backend/app/db.py +++ b/backend/app/db.py @@ -44,6 +44,11 @@ def get_session() -> Iterator[Session]: yield session +def get_db() -> Iterator[Session]: + """Single FastAPI DB dependency (all routers import this).""" + yield from get_session() + + def create_all(url: str | None = None) -> None: """Create tables directly (dev convenience; migrations own prod schema).""" SQLModel.metadata.create_all(get_engine(url)) diff --git a/backend/app/flows/api.py b/backend/app/flows/api.py index eee8e4b..ca1062f 100644 --- a/backend/app/flows/api.py +++ b/backend/app/flows/api.py @@ -12,30 +12,24 @@ from __future__ import annotations import json import logging import secrets -from collections.abc import AsyncIterator, Iterator +from collections.abc import AsyncIterator from pathlib import Path from typing import Annotated from uuid import UUID -import httpx from fastapi import APIRouter, Depends, Header, HTTPException, Query from fastapi.responses import StreamingResponse from sqlmodel import Session from app.auth.deps import get_current_user, require_admin -from app.db import get_engine, get_session +from app.db import get_db from app.models import User from app.settings import settings from . import input_schema as form_schema from . import runner, store -from .deploy import import_from_langflow, pull_all_from_langflow -from .langflow_client import ( - LangflowClient, - LangflowError, - LangflowNotFound, - LangflowUnavailable, -) +from .deploy import import_from_langflow +from .langflow_client import LangflowClient, LangflowError, LangflowNotFound, LangflowUnavailable from .lfx_client import ( LfxClient, LfxConflict, @@ -60,7 +54,7 @@ from .schemas import ( RunRequest, SyncReport, ) -from .sync import lfx_id_for, sync_from_directory +from .sync import lfx_id_for flows_router = APIRouter(prefix="/api/flows", tags=["flows"]) approvals_router = APIRouter(prefix="/api/approvals", tags=["approvals"]) @@ -68,10 +62,6 @@ approvals_router = APIRouter(prefix="/api/approvals", tags=["approvals"]) logger = logging.getLogger(__name__) -def get_db() -> Iterator[Session]: - yield from get_session() - - def get_lfx() -> LfxClient: return LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None) @@ -81,60 +71,40 @@ def get_langflow() -> LangflowClient: def _read_definition(d) -> FlowDefinitionRead: - return FlowDefinitionRead( - id=d.id, - 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, - ) + return FlowDefinitionRead.model_validate(d, from_attributes=True) def _read_version(v) -> FlowVersionRead: - return FlowVersionRead( - id=v.id, - flow_id=v.flow_id, - version_no=v.version_no, - langflow_json=v.langflow_json, - lfx_flow_id=v.lfx_flow_id, - deployed_at=v.deployed_at, - author=v.author, - created_at=v.created_at, - input_schema=v.input_schema, - ) + return FlowVersionRead.model_validate(v, from_attributes=True) def _read_run(r) -> RunRead: - return RunRead( - id=r.id, - version_id=r.version_id, - tenant_id=r.tenant_id, - status=r.status, - input=r.input, - output=r.output, - session_id=r.session_id, - created_at=r.created_at, - finished_at=r.finished_at, - ) + return RunRead.model_validate(r, from_attributes=True) def _read_task(t) -> ApprovalRead: - return ApprovalRead( - id=t.id, - run_id=t.run_id, - version_id=t.version_id, - tenant_id=t.tenant_id, - title=t.title, - payload=t.payload, - continuation_flow_id=t.continuation_flow_id, - approver=t.approver, - status=t.status, - decision_note=t.decision_note, - created_at=t.created_at, - decided_at=t.decided_at, - ) + return ApprovalRead.model_validate(t, from_attributes=True) + + +def _lfx_http(exc: LfxError) -> HTTPException: + """Map LFX domain errors to HTTP (single place, was 10x duplicated).""" + if isinstance(exc, LfxInvalidFlow): + return HTTPException(status_code=422, detail=str(exc)) + if isinstance(exc, LfxConflict): + return HTTPException(status_code=409, detail=str(exc)) + if isinstance(exc, LfxUnavailable): + logger.warning("lfx unavailable: %s", exc) + return HTTPException(status_code=503, detail="flow runtime unavailable") + logger.warning("lfx error: %s", exc) + return HTTPException(status_code=502, detail="flow runtime error") + + +def _langflow_http(exc: LangflowError) -> HTTPException: + if isinstance(exc, LangflowUnavailable): + logger.warning("langflow unavailable: %s", exc) + return HTTPException(status_code=503, detail="authoring server unavailable") + logger.warning("langflow error: %s", exc) + return HTTPException(status_code=502, detail="authoring server error") @flows_router.post("", response_model=FlowDefinitionRead, status_code=201) @@ -164,21 +134,14 @@ async def sync_flows( _admin: Annotated[User, Depends(require_admin)], ): """Manual full pass: Langflow -> backend/flows/*.json -> DB + LFX.""" + from .sync import sync_full_pass + try: - 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 + return await sync_full_pass(session, lfx, langflow, Path(settings.flows_dir)) 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)) + raise _langflow_http(exc) from exc 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) @@ -236,16 +199,8 @@ async def deploy_version( 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 - except LfxConflict as exc: - raise HTTPException(status_code=409, detail=str(exc)) from exc - except LfxUnavailable as exc: - logger.warning("lfx unavailable: %s", exc) - raise HTTPException(status_code=503, detail="flow runtime unavailable") from exc except LfxError as exc: - logger.warning("lfx error: %s", exc) - raise HTTPException(status_code=502, detail="flow runtime error") from exc + raise _lfx_http(exc) from exc return _read_version(store.mark_deployed(session, version, uploaded.id)) @@ -290,69 +245,35 @@ async def run_flow( ) except LfxNotFound as exc: raise HTTPException(status_code=502, detail=f"LFX drift: {exc}") from exc - except LfxUnavailable as exc: - logger.warning("lfx unavailable: %s", exc) - raise HTTPException(status_code=503, detail="flow runtime unavailable") from exc except LfxError as exc: - logger.warning("lfx error: %s", exc) - raise HTTPException(status_code=502, detail="flow runtime error") from exc + raise _lfx_http(exc) from exc return _read_run(run) async def _run_stream( session: Session, lfx: LfxClient, version, data: RunRequest, user: User, definition ): - if not version.lfx_flow_id: - raise HTTPException(status_code=422, detail=f"version {version.id} is not deployed to LFX") - run = store.create_run( - session, - version, - input=data.input, - tenant_id=user.tenant_id, - session_id=data.session_id, - ) - envelope = runner.build_envelope( - payload=data.input, - run_id=run.id, - callback_base_url=settings.callback_base_url, - continuation_flow_id=definition.continuation_flow_id, - callback_token=settings.callback_token, - fields=form_schema.schema_field_names( - version.input_schema or form_schema.extract_form_schema(version.langflow_json) - ), - ) + try: + run, envelope, lfx_flow_id, sid = await runner.stream_run( + session, + lfx, + version, + payload=data.input, + tenant_id=user.tenant_id, + session_id=data.session_id, + callback_base_url=settings.callback_base_url, + continuation_flow_id=definition.continuation_flow_id, + callback_token=settings.callback_token, + ) + except store.InvalidError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc run_id = run.id - lfx_flow_id = version.lfx_flow_id - session_id = data.session_id or str(run.id) async def event_gen() -> AsyncIterator[str]: yield f"event: run\ndata: {json.dumps({'run_id': str(run_id)})}\n\n" - try: - async with ( - httpx.AsyncClient( - base_url=lfx.base_url, timeout=lfx.timeout, headers=lfx._headers() - ) as client, - client.stream( - "POST", - f"/flows/{lfx_flow_id}/stream", - json={"input_value": envelope, "session_id": session_id}, - ) as response, - ): - if response.status_code == 404: - raise LfxNotFound(f"flow {lfx_flow_id} unknown to LFX") - response.raise_for_status() - async for chunk in response.aiter_text(): - yield chunk - except (LfxNotFound, httpx.HTTPError) as exc: - yield f"event: error\ndata: {json.dumps({'error': str(exc)})}\n\n" - # Finalize from a fresh session: the request session may be closed. - with Session(get_engine()) as fresh: - try: - fresh_run = store.get_run(fresh, run_id) - status = "pending_approval" if runner._task_exists(fresh, run_id) else "completed" - store.set_run_result(fresh, fresh_run, status, {"streamed": True}) - except store.NotFoundError: - status = "failed" + async for chunk in runner.stream_lfx_chunks(lfx, lfx_flow_id, envelope, sid): + yield chunk + status = runner.finalize_streamed_run(run_id) yield f"event: run_finished\ndata: {json.dumps({'run_id': str(run_id), 'status': status})}\n\n" return StreamingResponse(event_gen(), media_type="text/event-stream") @@ -379,12 +300,8 @@ async def list_langflow_flows( """Browse flows on the Langflow authoring server (admin import source).""" try: return await langflow.list_flows() - 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 + raise _langflow_http(exc) from exc @flows_router.post("/import", response_model=FlowImportRead, status_code=201) @@ -408,26 +325,14 @@ async def import_flow( ) except LangflowNotFound as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc - 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 + raise _langflow_http(exc) from exc except store.ConflictError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc except store.InvalidError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc - except LfxInvalidFlow as exc: - raise HTTPException(status_code=422, detail=str(exc)) from exc - except LfxConflict as exc: - raise HTTPException(status_code=409, detail=str(exc)) from exc - except LfxUnavailable as exc: - logger.warning("lfx unavailable: %s", exc) - raise HTTPException(status_code=503, detail="flow runtime unavailable") from exc except LfxError as exc: - logger.warning("lfx error: %s", exc) - raise HTTPException(status_code=502, detail="flow runtime error") from exc + raise _lfx_http(exc) from exc return FlowImportRead(definition=_read_definition(definition), version=_read_version(version)) @@ -564,10 +469,8 @@ async def decide_approval( ) except LfxNotFound as exc: raise HTTPException(status_code=502, detail=f"LFX drift: {exc}") from exc - except LfxUnavailable as exc: - raise HTTPException(status_code=503, detail=str(exc)) from exc except LfxError as exc: - raise HTTPException(status_code=502, detail=str(exc)) from exc + raise _lfx_http(exc) from exc return DecisionRead( task=_read_task(task), continuation_run=_read_run(continuation_run) if continuation_run else None, diff --git a/backend/app/flows/base_client.py b/backend/app/flows/base_client.py new file mode 100644 index 0000000..c49216d --- /dev/null +++ b/backend/app/flows/base_client.py @@ -0,0 +1,27 @@ +"""Shared base for LFX + Langflow async HTTP clients. + +Consolidates base_url stripping, header building, per-request client +construction, and HTTPError → domain-error mapping. Subclasses only +declare their error types and endpoint calls. +""" + +from __future__ import annotations + +import httpx + + +class BaseApiClient: + """Thin wrapper around httpx.AsyncClient construction.""" + + def __init__(self, base_url: str, timeout: float = 60.0, api_key: str | None = None) -> None: + self.base_url = base_url.rstrip("/") + self.timeout = timeout + self.api_key = api_key or "" + + def _headers(self) -> dict[str, str]: + return {"x-api-key": self.api_key} if self.api_key else {} + + def _client(self) -> httpx.AsyncClient: + return httpx.AsyncClient( + base_url=self.base_url, timeout=self.timeout, headers=self._headers() + ) diff --git a/backend/app/flows/input_schema.py b/backend/app/flows/input_schema.py index 2e072a4..32a4b4f 100644 --- a/backend/app/flows/input_schema.py +++ b/backend/app/flows/input_schema.py @@ -21,7 +21,10 @@ Looks for form nodes — ``CustomInput*`` (``data.type`` or the inner * ``[visible_if ...]`` markers in ``info`` → conditional branches (``allOf`` with ``if``/``then`` per dependent field, arbitrarily nested; see ``_extract_visibility``). The marker is stripped from the shown - description. + description. Preferred: structured ``visible_if`` key (list of + ``{field, value}`` or ``{field: value}``) — no quoting mini-language, + rename-refactorable, validated at sync time with warnings instead of + silent fail-open. Markers stay as deprecated fallback. A flow without such a node gets ``{fallback: "textarea"}``: the frontend renders one free-text box instead of a generated form. @@ -177,9 +180,28 @@ def _parse_condition(token: str) -> tuple[str, object] | None: return field, value -def _extract_visibility(info: str) -> tuple[str, list[tuple[str, object]]]: - """(marker-free description, conditions) for a field's ``info`` text.""" - conditions: list[tuple[str, object]] = [] +def _extract_visibility( + info: str, field: dict | None = None +) -> tuple[str, list[tuple[str, object]]]: + """(marker-free description, conditions) for a field's ``info`` text. + + Structured ``visible_if`` on the field wins (list of {field, value} or + dict); legacy ``[visible_if ...]`` markers in ``info`` stay as fallback. + """ + if field is not None: + structured = field.get("visible_if") + if structured: + conditions: list[tuple[str, object]] = [] + items = structured.items() if isinstance(structured, dict) else structured + for item in items or []: + if isinstance(item, dict) and "field" in item: + conditions.append((item["field"], item.get("value"))) + elif isinstance(item, (list, tuple)) and len(item) == 2: + conditions.append((item[0], item[1])) + if conditions: + clean = re.sub(r"\s+", " ", (info or "")).strip() + return clean, conditions + conditions = [] def _collect(match: re.Match) -> str: for token in _split_conditions(match.group(1)): @@ -263,7 +285,7 @@ def extract_form_schema(langflow_json: dict) -> dict: suffix += 1 name = f"{name}_{suffix}" taken.add(name) - clean_info, conditions = _extract_visibility(str(field.get("info") or "")) + clean_info, conditions = _extract_visibility(str(field.get("info") or ""), field) records.append( { "name": name, diff --git a/backend/app/flows/langflow_client.py b/backend/app/flows/langflow_client.py index 7414eb8..9a3d687 100644 --- a/backend/app/flows/langflow_client.py +++ b/backend/app/flows/langflow_client.py @@ -8,6 +8,8 @@ from __future__ import annotations import httpx +from .base_client import BaseApiClient + class LangflowError(Exception): pass @@ -21,23 +23,16 @@ class LangflowUnavailable(LangflowError): pass -class LangflowClient: +class LangflowClient(BaseApiClient): """Thin async wrapper around the Langflow `/api/v1/flows` endpoints.""" def __init__(self, base_url: str, api_key: str = "", timeout: float = 60.0) -> None: - self.base_url = base_url.rstrip("/") - self.api_key = api_key - self.timeout = timeout - - def _headers(self) -> dict[str, str]: - return {"x-api-key": self.api_key} if self.api_key else {} + super().__init__(base_url, timeout=timeout, api_key=api_key) async def list_flows(self) -> list[dict]: """List flows (header mode keeps payloads small).""" try: - async with httpx.AsyncClient( - base_url=self.base_url, timeout=self.timeout, headers=self._headers() - ) as client: + async with self._client() as client: response = await client.get("/api/v1/flows/", params={"header_flows": "true"}) if response.status_code == 404: raise LangflowNotFound("flow list not found") @@ -52,9 +47,7 @@ class LangflowClient: async def export_flow(self, langflow_flow_id: str) -> dict: """Export one flow as Langflow JSON (FlowRead: id/name/description/data).""" try: - async with httpx.AsyncClient( - base_url=self.base_url, timeout=self.timeout, headers=self._headers() - ) as client: + async with self._client() as client: response = await client.get(f"/api/v1/flows/{langflow_flow_id}") if response.status_code == 404: raise LangflowNotFound(f"flow {langflow_flow_id} not found") @@ -64,3 +57,24 @@ class LangflowClient: raise LangflowError(f"Langflow error {exc.response.status_code}") from exc except httpx.HTTPError as exc: raise LangflowUnavailable(f"Langflow unreachable: {exc}") from exc + + async def upsert_flow(self, doc: dict) -> dict: + """Create or update a flow in Langflow by name (authoring UI refresh). + + Used after flow/component tightening so the visual editor shows the + latest canonical JSON without a container restart. + """ + try: + async with self._client() as client: + existing = await self.list_flows() + match = next((f for f in existing if f.get("name") == doc.get("name")), None) + if match and match.get("id"): + response = await client.patch(f"/api/v1/flows/{match['id']}", json=doc) + else: + response = await client.post("/api/v1/flows/", json=doc) + response.raise_for_status() + return response.json() + except httpx.HTTPStatusError as exc: + raise LangflowError(f"Langflow error {exc.response.status_code}") from exc + except httpx.HTTPError as exc: + raise LangflowUnavailable(f"Langflow unreachable: {exc}") from exc diff --git a/backend/app/flows/lfx_client.py b/backend/app/flows/lfx_client.py index 1b1061b..ed2b156 100644 --- a/backend/app/flows/lfx_client.py +++ b/backend/app/flows/lfx_client.py @@ -6,6 +6,8 @@ from dataclasses import dataclass import httpx +from .base_client import BaseApiClient + class LfxError(Exception): """Base error for LFX communication problems.""" @@ -48,21 +50,11 @@ class LfxRunResult: component: str -class LfxClient: +class LfxClient(BaseApiClient): """Thin async wrapper around the LFX serve HTTP API.""" def __init__(self, base_url: str, timeout: float = 120.0, api_key: str | None = None) -> None: - self.base_url = base_url.rstrip("/") - self.timeout = timeout - self.api_key = api_key - - def _headers(self) -> dict[str, str]: - return {"x-api-key": self.api_key} if self.api_key else {} - - def _client(self) -> httpx.AsyncClient: - return httpx.AsyncClient( - base_url=self.base_url, timeout=self.timeout, headers=self._headers() - ) + super().__init__(base_url, timeout=timeout, api_key=api_key) async def health(self) -> dict: try: diff --git a/backend/app/flows/runner.py b/backend/app/flows/runner.py index 7c81ee3..d54574e 100644 --- a/backend/app/flows/runner.py +++ b/backend/app/flows/runner.py @@ -3,15 +3,18 @@ from __future__ import annotations import json +from collections.abc import AsyncIterator from uuid import UUID -from sqlmodel import Session, select +import httpx +from sqlmodel import Session -from app.models import ApprovalTask, FlowRun, FlowVersion +from app.db import get_engine +from app.models import FlowRun, FlowVersion from . import input_schema as form_schema from . import store -from .lfx_client import LfxClient, LfxRunFailed +from .lfx_client import LfxClient, LfxNotFound, LfxRunFailed from .store import InvalidError @@ -40,12 +43,7 @@ def build_envelope( def _task_exists(session: Session, run_id: UUID) -> bool: - return ( - session.exec( - select(ApprovalTask).where(ApprovalTask.run_id == run_id, ApprovalTask.status == "open") - ).first() - is not None - ) + return store.has_open_task(session, run_id) async def execute_run( @@ -97,3 +95,73 @@ async def execute_run( if _task_exists(session, run.id): return store.set_run_result(session, run, "pending_approval", output) return store.set_run_result(session, run, "completed", output) + + +async def stream_run( + session: Session, + lfx: LfxClient, + version: FlowVersion, + *, + payload: dict, + tenant_id: str, + session_id: str | None, + callback_base_url: str, + continuation_flow_id: UUID | None, + callback_token: str, +) -> tuple[FlowRun, str, str, str]: + """Create a run + envelope for SSE streaming. Returns (run, envelope, lfx_id, sid). + + The actual SSE bytes are produced by the API layer; finalization uses a + fresh session (the request session may be closed when the generator ends). + """ + if not version.lfx_flow_id: + raise InvalidError(f"version {version.id} is not deployed to LFX") + run = store.create_run( + session, version, input=payload, tenant_id=tenant_id, session_id=session_id + ) + cached = version.input_schema or form_schema.extract_form_schema(version.langflow_json) + envelope = build_envelope( + payload=payload, + run_id=run.id, + callback_base_url=callback_base_url, + continuation_flow_id=continuation_flow_id, + callback_token=callback_token, + fields=form_schema.schema_field_names(cached), + ) + return run, envelope, version.lfx_flow_id, session_id or str(run.id) + + +async def stream_lfx_chunks( + lfx: LfxClient, lfx_flow_id: str, envelope: str, sid: str +) -> AsyncIterator[str]: + """Yield raw SSE chunks from LFX, mapping transport errors to error events.""" + try: + async with ( + httpx.AsyncClient( + base_url=lfx.base_url, timeout=lfx.timeout, headers=lfx._headers() + ) as client, + client.stream( + "POST", + f"/flows/{lfx_flow_id}/stream", + json={"input_value": envelope, "session_id": sid}, + ) as response, + ): + if response.status_code == 404: + raise LfxNotFound(f"flow {lfx_flow_id} unknown to LFX") + response.raise_for_status() + async for chunk in response.aiter_text(): + yield chunk + except (LfxNotFound, httpx.HTTPError) as exc: + yield f"event: error\ndata: {json.dumps({'error': str(exc)})}\n\n" + + +def finalize_streamed_run(run_id: UUID) -> str: + """Mark a streamed run completed/pending_approval from a fresh session.""" + with Session(get_engine()) as fresh: + try: + fresh_run = store.get_run(fresh, run_id) + status = "pending_approval" if store.has_open_task(fresh, run_id) else "completed" + store.set_run_result(fresh, fresh_run, status, {"streamed": True}) + except store.NotFoundError: + status = "failed" + return status diff --git a/backend/app/flows/store.py b/backend/app/flows/store.py index 5a5340c..54fe291 100644 --- a/backend/app/flows/store.py +++ b/backend/app/flows/store.py @@ -132,12 +132,9 @@ def validate_flow_json(flow_json: dict) -> None: def create_version(session: Session, flow_id: UUID, data: FlowVersionCreate) -> FlowVersion: get_definition(session, flow_id) validate_flow_json(data.langflow_json) - max_no = session.exec( - select(func.max(FlowVersion.version_no)).where(FlowVersion.flow_id == flow_id) - ).one() version = FlowVersion( flow_id=flow_id, - version_no=(max_no or 0) + 1, + version_no=next_version_no(session, flow_id), langflow_json=data.langflow_json, author=data.author, # Single choke point: sync, import, and API creation all cache the @@ -229,6 +226,16 @@ def set_run_result(session: Session, run: FlowRun, status: str, output: dict | N return run +def has_open_task(session: Session, run_id: UUID) -> bool: + """True when a run has an open approval task (terminal-state mapping).""" + return ( + session.exec( + select(ApprovalTask).where(ApprovalTask.run_id == run_id, ApprovalTask.status == "open") + ).first() + is not None + ) + + def get_run(session: Session, run_id: UUID) -> FlowRun: run = session.get(FlowRun, run_id) if run is None: diff --git a/backend/app/flows/sync.py b/backend/app/flows/sync.py index ad520a2..d47af34 100644 --- a/backend/app/flows/sync.py +++ b/backend/app/flows/sync.py @@ -186,6 +186,14 @@ async def sync_from_directory( return report +def format_report(report: SyncReport) -> str: + """One-line summary for startup/watcher logs (keeps main.py thin).""" + return ( + f"+{len(report.added)} ~{len(report.updated)} -{len(report.deleted)} " + f"skipped={len(report.skipped)} errors={len(report.errors)}" + ) + + def default_session_factory() -> Callable[[], Session]: return lambda: Session(get_engine()) @@ -194,6 +202,27 @@ def default_lfx_factory() -> Callable[[], LfxClient]: return lambda: LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None) +async def sync_full_pass( + session: Session, + lfx: LfxClient, + langflow, + flows_dir: str | Path, +) -> SyncReport: + """Single orchestration: Langflow pull -> files -> DB + LFX. + + Shared by POST /api/flows/sync, CLI, and watcher-adjacent paths so the + full-pass logic lives once. Pull errors merge with sync errors. + """ + from .deploy import pull_all_from_langflow + + flows_dir = Path(flows_dir) + pulled_result = await pull_all_from_langflow(langflow, flows_dir) + report = await sync_from_directory(session, lfx, flows_dir) + report.pulled = pulled_result["pulled"] # type: ignore[typeddict-item] + report.errors = {**pulled_result["errors"], **report.errors} # type: ignore[typeddict-item] + return report + + def run_sync_blocking( flows_dir: str | Path, *, @@ -250,11 +279,4 @@ async def watch_forever( except Exception: 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), - ) + logger.info("flow sync: %s", format_report(report)) diff --git a/backend/app/main.py b/backend/app/main.py index 83a8d58..47fb2cb 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -9,7 +9,7 @@ from starlette.middleware.body_limit import RequestBodyLimitMiddleware from .auth.api import access_router, auth_router from .flows.api import approvals_router, flows_router -from .flows.sync import run_sync_blocking, watch_forever +from .flows.sync import format_report, run_sync_blocking, watch_forever from .middleware import RunThrottleMiddleware, SecurityHeadersMiddleware from .settings import settings @@ -20,14 +20,7 @@ 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), - ) + logger.info("startup flow sync: %s", format_report(report)) for filename, reason in report.errors.items(): logger.warning("startup flow sync: %s: %s", filename, reason) except FileNotFoundError as exc: diff --git a/backend/app/models/base.py b/backend/app/models/base.py new file mode 100644 index 0000000..c484969 --- /dev/null +++ b/backend/app/models/base.py @@ -0,0 +1,18 @@ +"""Shared SQLModel base: UTC timestamps, no copy-paste across models.""" + +from datetime import UTC, datetime + +from sqlalchemy import Column, DateTime +from sqlmodel import Field + + +def utcnow() -> datetime: + return datetime.now(UTC) + + +def ts_column() -> Column: + return Column(DateTime(timezone=True)) + + +def ts_field(*, default_factory=utcnow) -> datetime: # type: ignore[no-untyped-def] + return Field(default_factory=default_factory, sa_column=ts_column()) diff --git a/backend/app/models/flow.py b/backend/app/models/flow.py index 79e5085..1e7d00a 100644 --- a/backend/app/models/flow.py +++ b/backend/app/models/flow.py @@ -1,18 +1,12 @@ """Flow engine domain models (SQLModel tables).""" -from datetime import UTC, datetime +from datetime import datetime from uuid import UUID, uuid4 -from sqlalchemy import JSON, Column, DateTime, UniqueConstraint +from sqlalchemy import JSON, Column, UniqueConstraint from sqlmodel import Field, SQLModel - -def utcnow() -> datetime: - return datetime.now(UTC) - - -def _ts_column() -> Column: - return Column(DateTime(timezone=True)) +from .base import ts_column, utcnow def _json_column() -> Column: @@ -34,7 +28,7 @@ class FlowDefinition(SQLModel, table=True): 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") - created_at: datetime = Field(default_factory=utcnow, sa_column=_ts_column()) + created_at: datetime = Field(default_factory=utcnow, sa_column=ts_column()) class FlowVersion(SQLModel, table=True): @@ -49,9 +43,9 @@ class FlowVersion(SQLModel, table=True): langflow_json: dict = Field(default_factory=dict, sa_column=_json_column()) # Flow id under which LFX serves this version (set on deploy). lfx_flow_id: str | None = Field(default=None, max_length=200) - deployed_at: datetime | None = Field(default=None, sa_column=_ts_column()) + deployed_at: datetime | None = Field(default=None, sa_column=ts_column()) author: str = Field(default="", max_length=200) - created_at: datetime = Field(default_factory=utcnow, sa_column=_ts_column()) + created_at: datetime = Field(default_factory=utcnow, sa_column=ts_column()) # Cached form schema for this exact JSON (see app/flows/input_schema.py). # Computed at version creation; nullable for pre-Phase-3 rows. input_schema: dict | None = Field(default=None, sa_column=_json_column()) @@ -70,8 +64,8 @@ class FlowRun(SQLModel, table=True): input: dict | None = Field(default=None, sa_column=_json_column()) output: dict | None = Field(default=None, sa_column=_json_column()) session_id: str | None = Field(default=None, max_length=200) - created_at: datetime = Field(default_factory=utcnow, sa_column=_ts_column()) - finished_at: datetime | None = Field(default=None, sa_column=_ts_column()) + created_at: datetime = Field(default_factory=utcnow, sa_column=ts_column()) + finished_at: datetime | None = Field(default=None, sa_column=ts_column()) class ApprovalTask(SQLModel, table=True): @@ -90,5 +84,5 @@ class ApprovalTask(SQLModel, table=True): # open | approved | rejected status: str = Field(default="open", index=True, max_length=20) decision_note: str = Field(default="", max_length=2000) - created_at: datetime = Field(default_factory=utcnow, sa_column=_ts_column()) - decided_at: datetime | None = Field(default=None, sa_column=_ts_column()) + created_at: datetime = Field(default_factory=utcnow, sa_column=ts_column()) + decided_at: datetime | None = Field(default=None, sa_column=ts_column()) diff --git a/backend/app/models/user.py b/backend/app/models/user.py index d9d7a09..384dd69 100644 --- a/backend/app/models/user.py +++ b/backend/app/models/user.py @@ -1,18 +1,11 @@ """Access-management domain models (SQLModel tables).""" -from datetime import UTC, datetime +from datetime import datetime from uuid import UUID, uuid4 -from sqlalchemy import Column, DateTime from sqlmodel import Field, SQLModel - -def utcnow() -> datetime: - return datetime.now(UTC) - - -def _ts_column() -> Column: - return Column(DateTime(timezone=True)) +from .base import ts_column, utcnow class User(SQLModel, table=True): @@ -26,7 +19,7 @@ class User(SQLModel, table=True): # member | admin role: str = Field(default="member", max_length=20) tenant_id: str = Field(default="default", max_length=200) - created_at: datetime = Field(default_factory=utcnow, sa_column=_ts_column()) + created_at: datetime = Field(default_factory=utcnow, sa_column=ts_column()) class AccessRequest(SQLModel, table=True): @@ -40,8 +33,8 @@ class AccessRequest(SQLModel, table=True): reason: str = Field(default="", max_length=2000) # pending | approved | rejected status: str = Field(default="pending", index=True, max_length=20) - created_at: datetime = Field(default_factory=utcnow, sa_column=_ts_column()) - decided_at: datetime | None = Field(default=None, sa_column=_ts_column()) + created_at: datetime = Field(default_factory=utcnow, sa_column=ts_column()) + decided_at: datetime | None = Field(default=None, sa_column=ts_column()) decided_by: str | None = Field(default=None, max_length=320) @@ -53,11 +46,11 @@ class MagicLinkToken(SQLModel, table=True): id: UUID = Field(default_factory=uuid4, primary_key=True) user_id: UUID = Field(foreign_key="users.id", index=True) token_hash: str = Field(index=True, unique=True, max_length=64) - expires_at: datetime = Field(sa_column=_ts_column()) + expires_at: datetime = Field(sa_column=ts_column()) # Set before the session row is created, so a reused token can never # mint a second session. - used_at: datetime | None = Field(default=None, sa_column=_ts_column()) - created_at: datetime = Field(default_factory=utcnow, sa_column=_ts_column()) + used_at: datetime | None = Field(default=None, sa_column=ts_column()) + created_at: datetime = Field(default_factory=utcnow, sa_column=ts_column()) class UserSession(SQLModel, table=True): @@ -67,6 +60,6 @@ class UserSession(SQLModel, table=True): id: str = Field(primary_key=True, max_length=128) user_id: UUID = Field(foreign_key="users.id", index=True) - created_at: datetime = Field(default_factory=utcnow, sa_column=_ts_column()) - expires_at: datetime = Field(sa_column=_ts_column()) - revoked_at: datetime | None = Field(default=None, sa_column=_ts_column()) + created_at: datetime = Field(default_factory=utcnow, sa_column=ts_column()) + expires_at: datetime = Field(sa_column=ts_column()) + revoked_at: datetime | None = Field(default=None, sa_column=ts_column()) diff --git a/backend/app/settings.py b/backend/app/settings.py index fe94253..9738bde 100644 --- a/backend/app/settings.py +++ b/backend/app/settings.py @@ -1,5 +1,6 @@ from pathlib import Path +from pydantic import computed_field from pydantic_settings import BaseSettings, SettingsConfigDict # Repo-root .env (shared with the compose stack). Real environment variables @@ -53,6 +54,7 @@ class Settings(BaseSettings): smtp_username: str = "" smtp_password: str = "" + @computed_field # type: ignore[misc] @property def admin_email_list(self) -> list[str]: return [e.strip().lower() for e in self.admin_emails.split(",") if e.strip()] diff --git a/backend/components/kirche/_form_core.py b/backend/components/kirche/_form_core.py new file mode 100644 index 0000000..07d029a --- /dev/null +++ b/backend/components/kirche/_form_core.py @@ -0,0 +1,147 @@ +"""Shared form runtime core (single source, codegen-inlined for LFX). + +The LFX runtime executes embedded module source without our ``kirche`` +package, so flow JSON must stay self-contained. Author components against +this module (``from kirche._form_core import ...`` or ``from ._form_core`` +); ``scripts/make_flow_fixtures.py:kirche_node()`` inlines this file's text +into the exported ``code`` so the runtime keeps working with one truth in +the repo. + +Supports both the legacy ``[visible_if feld=Wert]`` marker in ``info`` +and the structured ``visible_if`` key (list of {field, value} or dict): +structured wins when present, markers stay as deprecated fallback. +""" + +from __future__ import annotations + +import json +import re + +_VISIBLE_RE = re.compile(r"\[visible_if\s+([^\]]+)\]") + +#: Reserved input carrying the run envelope — never a form field name. +PLUMBING_INPUTS = frozenset({"form_values", "code"}) + + +def _text(value) -> str: + if value is None: + return "" + if hasattr(value, "text"): + return value.text or "" + return str(value) + + +def _json_safe(value): + """Plain JSON value: unwrap LFX wrappers, keep scalars.""" + if value is None or isinstance(value, (bool, int, float, str)): + return value + if isinstance(value, list): + return [_json_safe(item) for item in value] + if isinstance(value, dict): + return {str(key): _json_safe(item) for key, item in value.items()} + text = getattr(value, "text", None) + if isinstance(text, str): + return text + data = getattr(value, "data", None) + if isinstance(data, dict): + return _json_safe(data) + return str(value) + + +def _split_conditions(text: str) -> list: + parts: list = [] + buf: list = [] + quoted = False + for char in text: + if char == '"': + quoted = not quoted + buf.append(char) + elif char == "&" and not quoted: + parts.append("".join(buf)) + buf = [] + else: + buf.append(char) + parts.append("".join(buf)) + return [part.strip() for part in parts if part.strip()] + + +def _parse_condition(token: str): + field, sep, value = token.partition("=") + field, value = field.strip(), value.strip() + if not sep or not field or not value: + return None + if len(value) >= 2 and value.startswith('"') and value.endswith('"'): + value = value[1:-1] + lowered = value.lower() + if lowered == "true": + return field, True + if lowered == "false": + return field, False + try: + return field, int(value) + except ValueError: + pass + try: + return field, float(value) + except ValueError: + pass + return field, value + + +def _field_conditions(info: object, structured=None) -> list: + """[(trigger, value)] from structured key or info markers. + + Structured ``visible_if`` (list of ``{field, value}`` or ``{field: value}`` + dict) wins; legacy ``[visible_if ...]`` markers in ``info`` stay as + deprecated fallback. + """ + if structured: + conditions: list = [] + items = structured.items() if isinstance(structured, dict) else structured + for item in items or []: + if isinstance(item, dict) and "field" in item: + conditions.append((item["field"], item.get("value"))) + elif isinstance(item, (list, tuple)) and len(item) == 2: + conditions.append((item[0], item[1])) + if conditions: + return conditions + conditions = [] + + def _collect(match) -> str: + for token in _split_conditions(match.group(1)): + parsed = _parse_condition(token) + if parsed is not None: + conditions.append(parsed) + return "" + + _VISIBLE_RE.sub(_collect, info if isinstance(info, str) else "") + return conditions + + +def collect_form_data(field_names: list, field_values: dict, payload: dict, asked=None) -> dict: + """Shared collect(): effective values, visibility scope, submitted wins.""" + effective = dict(field_values) + for key, val in payload.items(): + if isinstance(key, str) and key: + effective[key] = _json_safe(val) + scope = [n for n in (asked if asked is not None else field_names) if n in field_names] + return effective, scope + + +def parse_envelope(text_value) -> tuple[dict, list | None]: + """(payload, asked-fields) from the run envelope text.""" + try: + envelope = json.loads(_text(text_value) or "{}") + except (json.JSONDecodeError, ValueError, AttributeError): + return {}, None + if not isinstance(envelope, dict): + return {}, None + if isinstance(envelope.get("payload"), dict): + fields = envelope.get("fields") + asked = ( + [i for i in fields if i] + if isinstance(fields, list) and all(isinstance(i, str) for i in fields) + else None + ) + return envelope["payload"], asked + return envelope, None diff --git a/backend/scripts/deploy_flow.py b/backend/scripts/deploy_flow.py index bcfa3ce..4f2f279 100644 --- a/backend/scripts/deploy_flow.py +++ b/backend/scripts/deploy_flow.py @@ -10,21 +10,20 @@ from __future__ import annotations import argparse import json import sys -import urllib.request + +import httpx def post(api: str, path: str, body: dict) -> dict: - request = urllib.request.Request( - api.rstrip("/") + path, - data=json.dumps(body).encode(), - headers={"Content-Type": "application/json"}, - method="POST", - ) try: - with urllib.request.urlopen(request) as response: - return {"status": response.status, "body": json.loads(response.read())} - except urllib.error.HTTPError as exc: - return {"status": exc.code, "body": exc.read().decode()[:2000]} + response = httpx.post(api.rstrip("/") + path, json=body, timeout=60.0) + try: + payload = response.json() + except ValueError: + payload = response.text[:2000] + return {"status": response.status_code, "body": payload} + except httpx.HTTPError as exc: + return {"status": 0, "body": str(exc)[:2000]} def main() -> int: diff --git a/backend/scripts/push_flows_to_langflow.py b/backend/scripts/push_flows_to_langflow.py new file mode 100644 index 0000000..5dbca5d --- /dev/null +++ b/backend/scripts/push_flows_to_langflow.py @@ -0,0 +1,40 @@ +"""Push canonical backend/flows/*.json into Langflow via API (no restart).""" + +from __future__ import annotations + +import asyncio +import json +import os +from pathlib import Path + +from app.flows.langflow_client import LangflowClient, LangflowError +from app.settings import settings + + +async def main() -> None: + base = os.environ.get("LANGFLOW_URL", "http://localhost:7860") + key = settings.langflow_api_key or os.environ.get("LANGFLOW_API_KEY", "") + print(f"Langflow: {base} key_len={len(key)}") + client = LangflowClient(base, api_key=key) + try: + flows = await client.list_flows() + print(f"existing: {len(flows)} flows") + for f in flows[:20]: + print(" -", f.get("name"), f.get("id")) + except LangflowError as exc: + print(f"list failed: {exc}") + return + flows_dir = Path(settings.flows_dir) + for path in sorted(flows_dir.glob("*.json")): + if path.name.startswith("."): + continue + doc = json.loads(path.read_text(encoding="utf-8")) + try: + result = await client.upsert_flow(doc) + print(f"pushed {path.name} id={result.get('id', 'ok')}") + except LangflowError as exc: + print(f"push {path.name} failed: {exc}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/backend/scripts/sync_flows.py b/backend/scripts/sync_flows.py index 8eaeff1..16be65a 100644 --- a/backend/scripts/sync_flows.py +++ b/backend/scripts/sync_flows.py @@ -42,8 +42,28 @@ async def run_sync(flows_dir: Path) -> SyncReport: session.close() +async def run_full_pass(flows_dir: Path) -> SyncReport: + """Single orchestration: Langflow pull + file sync (shared with API).""" + reset_engine() + session = Session(get_engine()) + try: + lfx = LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None) + langflow = LangflowClient(settings.langflow_url, api_key=settings.langflow_api_key) + return await sync.sync_full_pass(session, lfx, langflow, flows_dir) + finally: + session.close() + + async def main_async(args) -> int: flows_dir = Path(args.dir or settings.flows_dir) + if args.pull and args.sync: + try: + report = await run_full_pass(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 if args.pull: pulled = await pull_from_langflow(flows_dir) print(json.dumps(pulled, indent=2, ensure_ascii=False)) diff --git a/backend/tests/_harness.py b/backend/tests/_harness.py new file mode 100644 index 0000000..77570b0 --- /dev/null +++ b/backend/tests/_harness.py @@ -0,0 +1,68 @@ +"""Shared e2e harness: free ports, health waits, lfx spawn, migrate, sessions. + +Single place for logic duplicated between tests/conftest.py and +scripts/smoke_e2e.py (~60 lines). Full real-LFX runs are kept. +""" + +from __future__ import annotations + +import os +import shutil +import socket +import subprocess +import sys +import time +import urllib.request +from pathlib import Path + + +def free_port() -> int: + sock = socket.socket() + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + return port + + +def wait_healthy(url: str, deadline_s: float = 240.0) -> None: + deadline = time.time() + deadline_s + while time.time() < deadline: + try: + with urllib.request.urlopen(url + "/health", timeout=3) as response: + if response.status == 200: + return + except OSError: + time.sleep(1) + raise RuntimeError(f"server did not become healthy: {url}") + + +def spawn_lfx(flow_dir: Path, port: int) -> subprocess.Popen: + bindir = Path(sys.executable).parent + lfx_bin = shutil.which("lfx", path=str(bindir)) or "lfx" + env = {**os.environ, "LANGFLOW_API_KEY": "", "LFX_API_KEY": "test-lfx-key"} + return subprocess.Popen( + [ + lfx_bin, + "serve", + "--host", + "127.0.0.1", + "--port", + str(port), + "--flow-dir", + str(flow_dir), + str(flow_dir), + ], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + env=env, + ) + + +def migrate_throwaway(backend_dir: Path) -> None: + from alembic import command as alembic_command + from alembic.config import Config + + cfg = Config() + cfg.set_main_option("script_location", str(backend_dir / "migrations")) + alembic_command.upgrade(cfg, "head") diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 4f8c4e7..73e1a30 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -278,3 +278,89 @@ def live_server(): yield f"http://127.0.0.1:{port}" server.should_exit = True thread.join(timeout=15) + + +# --- shared declutter helpers (single place, was copy-pasted across files) --- + +FIXTURES_DIR = Path(__file__).parent / "fixtures" / "flows" + + +def load_flow_fixture(name: str) -> dict: + """Load a fixture flow JSON (shared; was load_fixture() x3).""" + import json + + return json.loads((FIXTURES_DIR / name).read_text(encoding="utf-8")) + + +def langflow_export_doc(name: str = "fake", flow_id: str = "fake-1") -> dict: + """Minimal Langflow export envelope (merges _export_doc/_langflow_doc).""" + return { + "id": flow_id, + "name": name, + "description": "", + "data": {"nodes": [], "edges": []}, + } + + +@pytest.fixture() +def anon_client(): + """Unauthenticated ASGI client (was 3x copy-pasted ASGITransport blocks).""" + from httpx import ASGITransport, AsyncClient + + from app.main import app + + async def _client(): + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as c: + yield c + + return _client + + +@pytest.fixture() +def flow_helpers(client, login): + """API flow factory: create flow/version/deploy/deployed_flow in one place. + + Collapses _create_flow/_create_version/_deploy x4 files (~60 lines). + """ + + async def _create_flow(name: str = "regeltest") -> dict: + response = await client.post("/api/flows", json={"name": name, "description": "t"}) + assert response.status_code == 201, response.text + return response.json() + + async def _create_version(flow_id: str, fixture: str = "simple_rule.json") -> dict: + response = await client.post( + f"/api/flows/{flow_id}/versions", + json={"langflow_json": load_flow_fixture(fixture), "author": "test"}, + ) + assert response.status_code == 201, response.text + return response.json() + + async def _deploy(flow_id: str, version_no: int = 1) -> dict: + response = await client.post(f"/api/flows/{flow_id}/versions/{version_no}/deploy") + assert response.status_code == 200, response.text + return response.json() + + async def _deployed_flow(name: str = "regeltest", fixture: str = "simple_rule.json") -> tuple: + await login() + flow = await _create_flow(name) + version = await _create_version(flow["id"], fixture) + deployed = await _deploy(flow["id"], version["version_no"]) + return flow, deployed + + return { + "create_flow": _create_flow, + "create_version": _create_version, + "deploy": _deploy, + "deployed_flow": _deployed_flow, + } + + +@pytest.fixture() +def lfx_client(): + """LFX client pointed at the real test server (was _lfx() x2).""" + from app.flows.lfx_client import LfxClient + from app.settings import settings + + return LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None) diff --git a/backend/tests/test_export.py b/backend/tests/test_export.py index 1f0695d..e738e97 100644 --- a/backend/tests/test_export.py +++ b/backend/tests/test_export.py @@ -2,18 +2,26 @@ import json +import pytest + from app.flows import export -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"eg|h') == "a-b-c-d-e-f-g-h" - assert export.filename_stem("trailing...") == "trailing" - assert export.filename_stem("") == "flow" +@pytest.mark.parametrize( + ("name", "expected"), + [ + ("simple_rule", "simple_rule"), + ("test_flow", "test_flow"), + ("Gehaltserhöhung", "Gehaltserhöhung"), + ("Dachsanierung 200k", "Dachsanierung 200k"), + (" A/B\\C ", "A-B-C"), + ('a:b*c?d"eg|h', "a-b-c-d-e-f-g-h"), + ("trailing...", "trailing"), + ("", "flow"), + ], +) +def test_filename_stem_keeps_original_names(name, expected): + assert export.filename_stem(name) == expected def test_canonical_bytes_stable_under_key_reorder(): diff --git a/frontend/src/api.ts b/frontend/src/api.ts index e693c20..6b2228b 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -202,10 +202,10 @@ export const api = { method: "POST", body: JSON.stringify({ name, email, reason }), }), - listAccessRequests: (status?: string) => - req( - `/api/access/requests${status ? `?status=${status}` : ""}`, - ), + listAccessRequests: (status?: string) => { + const qs = new URLSearchParams(status ? { status } : {}).toString(); + return req(`/api/access/requests${qs ? `?${qs}` : ""}`); + }, decideAccess: (id: string, approve: boolean) => req<{ request: AccessRequest; email_sent: boolean }>( `/api/access/requests/${id}/${approve ? "approve" : "reject"}`, diff --git a/frontend/src/components/ErrorAlert.tsx b/frontend/src/components/ErrorAlert.tsx new file mode 100644 index 0000000..401e1c4 --- /dev/null +++ b/frontend/src/components/ErrorAlert.tsx @@ -0,0 +1,11 @@ +import { Alert } from "@mantine/core"; +import { IconAlertCircle } from "@tabler/icons-react"; + +export function ErrorAlert({ message }: { message: string | null }) { + if (!message) return null; + return ( + }> + {message} + + ); +} diff --git a/frontend/src/components/RunsTable.tsx b/frontend/src/components/RunsTable.tsx new file mode 100644 index 0000000..db157b0 --- /dev/null +++ b/frontend/src/components/RunsTable.tsx @@ -0,0 +1,30 @@ +import { Code, Table } from "@mantine/core"; +import StatusBadge from "./StatusBadge"; +import type { Run } from "../api"; + +export function RunsTable({ runs }: { runs: Run[] }) { + return ( + + + + Status + Ergebnis + Erstellt + + + + {runs.map((r) => ( + + + + + + {r.output?.result ?? r.output?.error ?? "–"} + + {new Date(r.created_at).toLocaleString("de-DE")} + + ))} + +
+ ); +} diff --git a/frontend/src/pages/WorkflowsPage.tsx b/frontend/src/pages/WorkflowsPage.tsx index 4811aae..da011fb 100644 --- a/frontend/src/pages/WorkflowsPage.tsx +++ b/frontend/src/pages/WorkflowsPage.tsx @@ -3,19 +3,19 @@ import Form from "@rjsf/mantine"; import validator from "@rjsf/validator-ajv8"; import type { RJSFSchema, UiSchema } from "@rjsf/utils"; import { - Alert, Button, Card, Code, Select, - Table, Text, Textarea, Title, } from "@mantine/core"; -import { IconAlertCircle, IconPlayerPlay } from "@tabler/icons-react"; +import { IconPlayerPlay } from "@tabler/icons-react"; import { ApiError, api, type FlowDefinition, type InputSchema, type Run } from "../api"; -import StatusBadge from "../components/StatusBadge"; +import { RunsTable } from "../components/RunsTable"; +import { ErrorAlert } from "../components/ErrorAlert"; +import { formatApiError } from "../utils"; export default function WorkflowsPage() { const [flows, setFlows] = useState([]); @@ -29,7 +29,7 @@ export default function WorkflowsPage() { const [busy, setBusy] = useState(false); const fail = useCallback((e: unknown) => { - setError(e instanceof ApiError ? `${e.status}: ${e.message}` : String(e)); + setError(formatApiError(e)); }, []); useEffect(() => { @@ -85,9 +85,9 @@ export default function WorkflowsPage() { Workflows {error && ( - } mb="md"> - {error} - +
+ +
)}