From 5f3303fb21b548c20a2656c4143057df662e1e2a Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 29 Jul 2026 20:37:08 +0200 Subject: [PATCH] Add governed Dataflow search source --- src/govoplan_dataflow/backend/manifest.py | 22 ++ .../backend/search_source.py | 226 ++++++++++++++++++ tests/test_search_source.py | 179 ++++++++++++++ webui/src/features/dataflow/DataflowPage.tsx | 10 +- 4 files changed, 435 insertions(+), 2 deletions(-) create mode 100644 src/govoplan_dataflow/backend/search_source.py create mode 100644 tests/test_search_source.py diff --git a/src/govoplan_dataflow/backend/manifest.py b/src/govoplan_dataflow/backend/manifest.py index b7fd9d6..29a586e 100644 --- a/src/govoplan_dataflow/backend/manifest.py +++ b/src/govoplan_dataflow/backend/manifest.py @@ -35,6 +35,7 @@ from govoplan_core.core.datasources import ( from govoplan_core.core.policy import ( CAPABILITY_POLICY_DEFINITION_GOVERNANCE, ) +from govoplan_core.core.search import SearchSourceProviderRegistration from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base from govoplan_dataflow.backend.db import models as dataflow_models @@ -196,6 +197,14 @@ def _trigger_provider(context: ModuleContext): return SqlDataflowTriggerDispatcher(registry=context.registry) +def _search_source_provider(context: ModuleContext): + from govoplan_dataflow.backend.search_source import ( + create_dataflow_search_source, + ) + + return create_dataflow_search_source(context) + + def _tenant_summary(session, tenant_id: str) -> dict[str, int]: return { "dataflow_pipelines": ( @@ -241,6 +250,7 @@ manifest = ModuleManifest( "policy", "reporting", "risk_compliance", + "search", "workflow", ), optional_capabilities=( @@ -292,6 +302,12 @@ manifest = ModuleManifest( version_max_exclusive="1.0.0", optional=True, ), + ModuleInterfaceRequirement( + name="search.source", + version_min="1.0.0", + version_max_exclusive="2.0.0", + optional=True, + ), ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, @@ -339,6 +355,12 @@ manifest = ModuleManifest( CAPABILITY_DATAFLOW_RUN_LIFECYCLE: _run_provider, CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER: _trigger_provider, }, + search_sources=( + SearchSourceProviderRegistration( + id="dataflow.pipelines", + factory=_search_source_provider, + ), + ), tenant_summary_providers=(_tenant_summary,), migration_spec=MigrationSpec( module_id=MODULE_ID, diff --git a/src/govoplan_dataflow/backend/search_source.py b/src/govoplan_dataflow/backend/search_source.py new file mode 100644 index 0000000..d2f761b --- /dev/null +++ b/src/govoplan_dataflow/backend/search_source.py @@ -0,0 +1,226 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from urllib.parse import quote + +from sqlalchemy import func, or_, select +from sqlalchemy.orm import Session + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.modules import ModuleContext +from govoplan_core.core.search import ( + SearchAuthorizationRequest, + SearchBackfillPage, + SearchBackfillRequest, + SearchDocument, + SearchResourceType, +) +from govoplan_dataflow.backend.db.models import DataflowPipeline +from govoplan_dataflow.backend.governance import definition_decision + + +PROVIDER_ID = "dataflow.pipelines" +RESOURCE_TYPE = "dataflow_pipeline" +READ_SCOPE = "dataflow:pipeline:read" +ADMIN_SCOPE = "dataflow:pipeline:admin" + + +class DataflowSearchSource: + def __init__(self, registry: object | None) -> None: + self.registry = registry + + def resource_types(self) -> Sequence[SearchResourceType]: + return ( + SearchResourceType( + provider_id=PROVIDER_ID, + module_id="dataflow", + resource_type=RESOURCE_TYPE, + label="Data pipelines", + requires_authorization_recheck=True, + ), + ) + + def backfill( + self, + session: object, + *, + request: SearchBackfillRequest, + ) -> SearchBackfillPage: + db = _session(session) + if ( + request.provider_id != PROVIDER_ID + or request.resource_type != RESOURCE_TYPE + ): + raise ValueError("Unsupported Dataflow search source.") + + source_filter = or_( + DataflowPipeline.tenant_id == request.tenant_id, + DataflowPipeline.tenant_id.is_(None), + ) + statement = select(DataflowPipeline).where( + source_filter, + DataflowPipeline.deleted_at.is_(None), + ) + if request.cursor: + statement = statement.where( + DataflowPipeline.id > request.cursor + ) + rows = list( + db.scalars( + statement.order_by(DataflowPipeline.id).limit( + request.limit + 1 + ) + ) + ) + has_more = len(rows) > request.limit + selected = rows[: request.limit] + high_watermark = db.scalar( + select(func.max(DataflowPipeline.updated_at)).where( + source_filter, + DataflowPipeline.deleted_at.is_(None), + ) + ) + return SearchBackfillPage( + documents=tuple( + _pipeline_document( + pipeline, + tenant_id=request.tenant_id, + ) + for pipeline in selected + ), + next_cursor=selected[-1].id if has_more and selected else None, + complete=not has_more, + high_watermark=( + high_watermark.isoformat() + if high_watermark is not None + else None + ), + ) + + def authorize( + self, + session: object, + principal: object, + *, + requests: Sequence[SearchAuthorizationRequest], + ) -> Mapping[str, bool]: + decisions = { + request.reference.key: False + for request in requests + } + if not isinstance(principal, ApiPrincipal): + return decisions + if not ( + principal.has(READ_SCOPE) + or principal.has(ADMIN_SCOPE) + ): + return decisions + + valid_requests = tuple( + request + for request in requests + if ( + request.reference.tenant_id == principal.tenant_id + and request.reference.module_id == "dataflow" + and request.reference.resource_type == RESOURCE_TYPE + ) + ) + if not valid_requests: + return decisions + + db = _session(session) + ids = { + request.reference.resource_id + for request in valid_requests + } + pipelines = { + pipeline.id: pipeline + for pipeline in db.scalars( + select(DataflowPipeline).where( + DataflowPipeline.id.in_(ids), + or_( + DataflowPipeline.tenant_id + == principal.tenant_id, + DataflowPipeline.tenant_id.is_(None), + ), + DataflowPipeline.deleted_at.is_(None), + ) + ) + } + for request in valid_requests: + pipeline = pipelines.get( + request.reference.resource_id + ) + if pipeline is None: + continue + try: + allowed = definition_decision( + pipeline, + principal=principal, + registry=self.registry, + action="view", + ).allowed + except (PermissionError, RuntimeError, ValueError): + allowed = False + decisions[request.reference.key] = allowed + return decisions + + +def create_dataflow_search_source( + context: ModuleContext, +) -> DataflowSearchSource: + return DataflowSearchSource(context.registry) + + +def _pipeline_document( + pipeline: DataflowPipeline, + *, + tenant_id: str, +) -> SearchDocument: + updated_at = pipeline.updated_at + source_revision = ( + f"{pipeline.current_revision}:" + f"{updated_at.isoformat() if updated_at is not None else 'unknown'}" + ) + return SearchDocument( + tenant_id=tenant_id, + module_id="dataflow", + provider_id=PROVIDER_ID, + resource_type=RESOURCE_TYPE, + resource_id=pipeline.id, + title=pipeline.name, + url=f"/dataflow?pipelineId={quote(pipeline.id, safe='')}", + summary=pipeline.description, + body=pipeline.description, + keywords=( + pipeline.status, + pipeline.scope_type, + pipeline.definition_kind, + ), + visibility="tenant", + source_revision=source_revision, + source_updated_at=updated_at, + metadata={ + "status": pipeline.status, + "scope_type": pipeline.scope_type, + "definition_kind": pipeline.definition_kind, + "current_revision": pipeline.current_revision, + }, + requires_authorization_recheck=True, + ) + + +def _session(value: object) -> Session: + if not isinstance(value, Session): + raise TypeError( + "Dataflow search requires a SQLAlchemy session." + ) + return value + + +__all__ = [ + "DataflowSearchSource", + "PROVIDER_ID", + "RESOURCE_TYPE", + "create_dataflow_search_source", +] diff --git a/tests/test_search_source.py b/tests/test_search_source.py new file mode 100644 index 0000000..a3d3432 --- /dev/null +++ b/tests/test_search_source.py @@ -0,0 +1,179 @@ +from __future__ import annotations + +from types import SimpleNamespace +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.search import ( + SearchAuthorizationRequest, + SearchBackfillRequest, + SearchResourceReference, +) +from govoplan_core.db.base import Base +from govoplan_dataflow.backend.db.models import DataflowPipeline +from govoplan_dataflow.backend.manifest import ( + ADMIN_SCOPE, + READ_SCOPE, + get_manifest, +) +from govoplan_dataflow.backend.search_source import ( + DataflowSearchSource, + PROVIDER_ID, + RESOURCE_TYPE, +) + + +class DataflowSearchSourceTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite://") + Base.metadata.create_all( + self.engine, + tables=(DataflowPipeline.__table__,), + ) + self.session = Session(self.engine) + self.session.add_all( + ( + _pipeline( + "00000000-0000-0000-0000-000000000001", + "tenant-1", + "Monthly reconciliation", + ), + _pipeline( + "00000000-0000-0000-0000-000000000002", + "tenant-1", + "Sanctions screening", + ), + _pipeline( + "00000000-0000-0000-0000-000000000003", + "tenant-2", + "Other tenant", + ), + ) + ) + self.session.commit() + self.source = DataflowSearchSource(registry=None) + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def test_manifest_announces_optional_search_source(self) -> None: + manifest = get_manifest() + + self.assertIn("search", manifest.optional_dependencies) + self.assertEqual( + (PROVIDER_ID,), + tuple(item.id for item in manifest.search_sources), + ) + + def test_backfill_is_tenant_bounded_and_stably_paged(self) -> None: + first = self.source.backfill( + self.session, + request=SearchBackfillRequest( + tenant_id="tenant-1", + provider_id=PROVIDER_ID, + resource_type=RESOURCE_TYPE, + rebuild_id="rebuild-1", + limit=1, + ), + ) + second = self.source.backfill( + self.session, + request=SearchBackfillRequest( + tenant_id="tenant-1", + provider_id=PROVIDER_ID, + resource_type=RESOURCE_TYPE, + rebuild_id="rebuild-1", + cursor=first.next_cursor, + limit=1, + ), + ) + + self.assertFalse(first.complete) + self.assertTrue(second.complete) + self.assertEqual( + {"Monthly reconciliation", "Sanctions screening"}, + { + first.documents[0].title, + second.documents[0].title, + }, + ) + self.assertTrue( + first.documents[0].url.startswith( + "/dataflow?pipelineId=" + ) + ) + self.assertTrue( + first.documents[0].requires_authorization_recheck + ) + + def test_authorization_rechecks_tenant_scope_and_permission(self) -> None: + reference = SearchResourceReference( + tenant_id="tenant-1", + module_id="dataflow", + resource_type=RESOURCE_TYPE, + resource_id="00000000-0000-0000-0000-000000000001", + ) + request = SearchAuthorizationRequest( + reference=reference, + source_revision="1", + ) + + allowed = self.source.authorize( + self.session, + _principal({READ_SCOPE}), + requests=(request,), + ) + denied = self.source.authorize( + self.session, + _principal(set()), + requests=(request,), + ) + administrative = self.source.authorize( + self.session, + _principal({ADMIN_SCOPE}), + requests=(request,), + ) + + self.assertTrue(allowed[reference.key]) + self.assertFalse(denied[reference.key]) + self.assertTrue(administrative[reference.key]) + + +def _pipeline( + pipeline_id: str, + tenant_id: str, + name: str, +) -> DataflowPipeline: + return DataflowPipeline( + id=pipeline_id, + tenant_id=tenant_id, + scope_type="tenant", + scope_id=tenant_id, + definition_kind="flow", + name=name, + description=f"{name} description", + status="active", + current_revision=1, + ) + + +def _principal(scopes: set[str]) -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id="account-1", + membership_id="membership-1", + tenant_id="tenant-1", + scopes=frozenset(scopes), + ), + account=SimpleNamespace(id="account-1"), + user=SimpleNamespace(id="membership-1"), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/src/features/dataflow/DataflowPage.tsx b/webui/src/features/dataflow/DataflowPage.tsx index c9b28ab..75afa2a 100644 --- a/webui/src/features/dataflow/DataflowPage.tsx +++ b/webui/src/features/dataflow/DataflowPage.tsx @@ -43,6 +43,7 @@ import { type AuthInfo } from "@govoplan/core-webui"; import { ReactFlowProvider } from "@xyflow/react"; +import { useLocation } from "react-router-dom"; import { compileDataflowSql, createDataflowTrigger, @@ -99,6 +100,11 @@ const AUTO_PREVIEW_STORAGE_KEY = "govoplan.dataflow.auto-preview"; export default function DataflowPage({ settings, auth }: { settings: ApiSettings; auth: AuthInfo }) { const { requestNavigation, requestDiscard } = useUnsavedChanges(); + const location = useLocation(); + const requestedPipelineId = useMemo( + () => new URLSearchParams(location.search).get("pipelineId"), + [location.search] + ); const [pipelines, setPipelines] = useState([]); const [draft, setDraft] = useState(null); const [savedDraft, setSavedDraft] = useState(null); @@ -219,8 +225,8 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings }, [draft?.id, settings]); useEffect(() => { - void loadPipelines(); - }, []); + void loadPipelines(requestedPipelineId); + }, [requestedPipelineId]); useEffect(() => { let cancelled = false;