diff --git a/backend/app/flows/api.py b/backend/app/flows/api.py index 10eb03c..a18d4a9 100644 --- a/backend/app/flows/api.py +++ b/backend/app/flows/api.py @@ -15,6 +15,13 @@ from app.db import get_engine, get_session from app.settings import settings from . import runner, store +from .deploy import import_from_langflow +from .langflow_client import ( + LangflowClient, + LangflowError, + LangflowNotFound, + LangflowUnavailable, +) from .lfx_client import ( LfxClient, LfxConflict, @@ -29,6 +36,8 @@ from .schemas import ( ApprovalRead, FlowDefinitionCreate, FlowDefinitionRead, + FlowImportRead, + FlowImportRequest, FlowVersionCreate, FlowVersionRead, RunRead, @@ -47,6 +56,12 @@ def get_lfx() -> LfxClient: return LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None) +def get_langflow() -> LangflowClient: + return LangflowClient( + settings.langflow_url, api_key=settings.langflow_api_key or "" + ) + + def _read_definition(d) -> FlowDefinitionRead: return FlowDefinitionRead( id=d.id, @@ -282,6 +297,58 @@ def list_runs(flow_id: UUID, session: Session = Depends(get_db)): raise HTTPException(status_code=404, detail=str(exc)) from exc +@flows_router.get("/langflow/available") +async def list_langflow_flows(langflow: LangflowClient = Depends(get_langflow)): + """Browse flows on the Langflow authoring server (admin import source).""" + try: + return await langflow.list_flows() + except LangflowUnavailable as exc: + raise HTTPException(status_code=503, detail=str(exc)) from exc + except LangflowError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + + +@flows_router.post("/import", response_model=FlowImportRead, status_code=201) +async def import_flow( + data: FlowImportRequest, + session: Session = Depends(get_db), + langflow: LangflowClient = Depends(get_langflow), + lfx: LfxClient = Depends(get_lfx), +): + """Import a flow from Langflow: export -> version -> (deploy to LFX).""" + try: + definition, version = await import_from_langflow( + session, + langflow, + lfx, + data.langflow_flow_id, + name=data.name, + author=data.author, + deploy=data.deploy, + ) + except LangflowNotFound as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except LangflowUnavailable as exc: + raise HTTPException(status_code=503, detail=str(exc)) from exc + except LangflowError as exc: + raise HTTPException(status_code=502, detail=str(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: + raise HTTPException(status_code=503, detail=str(exc)) from exc + except LfxError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + return FlowImportRead( + definition=_read_definition(definition), version=_read_version(version) + ) + + @flows_router.get("/runs/{run_id}", response_model=RunRead) def get_run(run_id: UUID, session: Session = Depends(get_db)): try: diff --git a/backend/app/flows/deploy.py b/backend/app/flows/deploy.py new file mode 100644 index 0000000..2d6e961 --- /dev/null +++ b/backend/app/flows/deploy.py @@ -0,0 +1,60 @@ +"""Authoring pipeline: Langflow export -> versioned definition -> LFX deploy.""" + +from __future__ import annotations + +from sqlmodel import Session + +from app.models import FlowDefinition, FlowVersion + +from . import store +from .langflow_client import LangflowClient +from .lfx_client import LfxClient +from .schemas import FlowDefinitionCreate, FlowVersionCreate + + +async def import_from_langflow( + session: Session, + langflow: LangflowClient, + lfx: LfxClient, + langflow_flow_id: str, + *, + name: str | None = None, + author: str = "", + deploy: bool = True, +) -> tuple[FlowDefinition, FlowVersion]: + """Export a flow from Langflow, store it as a new version, deploy to LFX. + + Definitions are matched by name (explicit `name` wins, else the exported + name); every import creates a new immutable version. + """ + export = await langflow.export_flow(langflow_flow_id) + flow_name = name or export.get("name") or f"langflow-{langflow_flow_id}" + langflow_json = { + "id": export.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), + } + 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( + session, + definition.id, + FlowVersionCreate(langflow_json=langflow_json, author=author), + ) + 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) + return definition, version diff --git a/backend/app/flows/langflow_client.py b/backend/app/flows/langflow_client.py new file mode 100644 index 0000000..7414eb8 --- /dev/null +++ b/backend/app/flows/langflow_client.py @@ -0,0 +1,66 @@ +"""Async client for the Langflow authoring server (visual designer). + +Only used for the authoring pipeline (list/export flows). Execution always +goes through LFX (see lfx_client.py). +""" + +from __future__ import annotations + +import httpx + + +class LangflowError(Exception): + pass + + +class LangflowNotFound(LangflowError): + pass + + +class LangflowUnavailable(LangflowError): + pass + + +class LangflowClient: + """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 {} + + 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: + response = await client.get("/api/v1/flows/", params={"header_flows": "true"}) + if response.status_code == 404: + raise LangflowNotFound("flow list not found") + response.raise_for_status() + data = response.json() + return data if isinstance(data, list) else data.get("items", []) + 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 + + 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: + 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") + 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/schemas.py b/backend/app/flows/schemas.py index 01eaedf..84270b5 100644 --- a/backend/app/flows/schemas.py +++ b/backend/app/flows/schemas.py @@ -84,3 +84,17 @@ class ApprovalRead(BaseModel): decision_note: str created_at: datetime decided_at: datetime | None + + +class FlowImportRequest(BaseModel): + """Import a flow from the Langflow authoring server by its Langflow id.""" + + langflow_flow_id: str = Field(min_length=1) + name: str | None = Field(default=None, max_length=200) + author: str = Field(default="", max_length=200) + deploy: bool = True + + +class FlowImportRead(BaseModel): + definition: FlowDefinitionRead + version: FlowVersionRead diff --git a/backend/scripts/deploy_flow.py b/backend/scripts/deploy_flow.py new file mode 100644 index 0000000..bcfa3ce --- /dev/null +++ b/backend/scripts/deploy_flow.py @@ -0,0 +1,54 @@ +"""Import a flow from Langflow into our backend via the running API. + +Usage: + uv run python scripts/deploy_flow.py --api http://localhost:8000 \\ + --langflow-id [--name my-flow] [--author jane] [--no-deploy] +""" + +from __future__ import annotations + +import argparse +import json +import sys +import urllib.request + + +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]} + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--api", default="http://localhost:8000") + parser.add_argument("--langflow-id", required=True) + parser.add_argument("--name", default=None) + parser.add_argument("--author", default="") + parser.add_argument("--no-deploy", action="store_true") + args = parser.parse_args() + + result = post( + args.api, + "/api/flows/import", + { + "langflow_flow_id": args.langflow_id, + "name": args.name, + "author": args.author, + "deploy": not args.no_deploy, + }, + ) + print(json.dumps(result, indent=2, ensure_ascii=False)) + return 0 if result["status"] in (200, 201) else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index ef3e9e1..7629df3 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -166,6 +166,35 @@ def lfx_server(tmp_path_factory): proc.kill() +@pytest.fixture() +def override_langflow(): + """Replace the Langflow dependency with a fake (real LFX stays untouched). + + Usage: override_langflow(export_doc) -- the fake export_flow() returns the + doc, or raises LangflowNotFound/Unavailable for status=404/0. + """ + from app.flows.api import get_langflow + from app.flows.langflow_client import LangflowNotFound, LangflowUnavailable + from app.main import app + + def _override(export_doc=None, status=200): + class FakeLangflow: + async def export_flow(self, flow_id): + if status == 404: + raise LangflowNotFound(f"flow {flow_id} not found") + if status == 0: + raise LangflowUnavailable("Langflow unreachable") + return export_doc + + async def list_flows(self): + return [{"id": "fake-1", "name": "fake"}] + + app.dependency_overrides[get_langflow] = lambda: FakeLangflow() + + yield _override + app.dependency_overrides.pop(get_langflow, None) + + @pytest.fixture(scope="session") def live_server(): """Real uvicorn server (ephemeral port) so LFX flows can call back diff --git a/backend/tests/test_deploy.py b/backend/tests/test_deploy.py new file mode 100644 index 0000000..7af0843 --- /dev/null +++ b/backend/tests/test_deploy.py @@ -0,0 +1,109 @@ +"""Step 4 tests: Langflow import pipeline (Langflow faked, LFX real).""" + +import json +from pathlib import Path +from uuid import uuid4 + +FIXTURES = Path(__file__).parent / "fixtures" / "flows" +LANGFLOW_ID = "11111111-2222-4333-8444-555555555555" + + +def _export_doc(name="importiert"): + fixture = json.loads((FIXTURES / "simple_rule.json").read_text()) + return { + "id": LANGFLOW_ID, + "name": name, + "description": "aus Langflow", + "endpoint_name": None, + "data": fixture["data"], + "tags": [], + "webhook": False, + "is_component": False, + } + + +async def test_import_creates_definition_version_and_deploys( + client, lfx_server, override_langflow +): + override_langflow(_export_doc()) + response = await client.post( + "/api/flows/import", + json={"langflow_flow_id": LANGFLOW_ID, "author": "admin"}, + ) + assert response.status_code == 201, response.text + body = response.json() + 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"]["deployed_at"] is not None + assert body["version"]["langflow_json"]["data"]["nodes"] + + # Imported flow actually runs on LFX. + definition_id = body["definition"]["id"] + response = await client.post( + f"/api/flows/{definition_id}/run", json={"input": {"betrag": 10}} + ) + assert response.status_code == 200, response.text + assert response.json()["status"] == "completed" + + +async def test_import_reuses_definition_and_bumps_version( + client, lfx_server, override_langflow +): + override_langflow(_export_doc()) + first = await client.post( + "/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID} + ) + second = await client.post( + "/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID} + ) + assert first.status_code == 201 + assert second.status_code == 201 + assert first.json()["definition"]["id"] == second.json()["definition"]["id"] + assert second.json()["version"]["version_no"] == 2 + + +async def test_import_with_explicit_name(client, lfx_server, override_langflow): + override_langflow(_export_doc(name="langflow-name")) + response = await client.post( + "/api/flows/import", + json={"langflow_flow_id": LANGFLOW_ID, "name": "wunschname"}, + ) + assert response.status_code == 201, response.text + assert response.json()["definition"]["name"] == "wunschname" + + +async def test_import_without_deploy(client, override_langflow): + override_langflow(_export_doc()) + response = await client.post( + "/api/flows/import", + json={"langflow_flow_id": LANGFLOW_ID, "deploy": False}, + ) + assert response.status_code == 201, response.text + body = response.json() + assert body["version"]["lfx_flow_id"] is None + assert body["version"]["deployed_at"] is None + + +async def test_import_langflow_not_found(client, override_langflow): + override_langflow(status=404) + response = await client.post( + "/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID} + ) + assert response.status_code == 404 + + +async def test_import_langflow_unreachable(client, override_langflow): + override_langflow(status=0) + response = await client.post( + "/api/flows/import", json={"langflow_flow_id": str(uuid4())} + ) + assert response.status_code == 503 + + +async def test_list_langflow_flows(client, override_langflow): + override_langflow(_export_doc()) + response = await client.get("/api/flows/langflow/available") + assert response.status_code == 200 + assert response.json() == [{"id": "fake-1", "name": "fake"}] diff --git a/backend/tests/test_langflow_client.py b/backend/tests/test_langflow_client.py new file mode 100644 index 0000000..ad93ee4 --- /dev/null +++ b/backend/tests/test_langflow_client.py @@ -0,0 +1,52 @@ +"""LangflowClient HTTP mapping tests (Langflow mocked via respx, no LFX).""" + +import httpx +import pytest +import respx + +from app.flows.langflow_client import ( + LangflowClient, + LangflowNotFound, + LangflowUnavailable, +) + +BASE = "https://langflow.test" + + +def _client(): + return LangflowClient(BASE, api_key="test-key") + + +@respx.mock +async def test_export_flow_sends_api_key_and_returns_doc(): + doc = {"id": "abc", "name": "n", "data": {"nodes": [], "edges": []}} + route = respx.get(f"{BASE}/api/v1/flows/abc").mock( + return_value=httpx.Response(200, json=doc) + ) + result = await _client().export_flow("abc") + assert result == doc + assert route.called + assert route.calls[0].request.headers["x-api-key"] == "test-key" + + +@respx.mock +async def test_export_flow_not_found(): + respx.get(f"{BASE}/api/v1/flows/missing").mock( + return_value=httpx.Response(404, json={"detail": "nope"}) + ) + with pytest.raises(LangflowNotFound): + await _client().export_flow("missing") + + +@respx.mock +async def test_list_flows_unwraps_items(): + respx.get(f"{BASE}/api/v1/flows/").mock( + return_value=httpx.Response(200, json={"items": [{"id": "1"}]}) + ) + assert await _client().list_flows() == [{"id": "1"}] + + +async def test_unreachable_raises_unavailable(): + client = LangflowClient("https://langflow.dead", api_key="k") + with pytest.raises(LangflowUnavailable): + await client.export_flow("whatever")