From baa2143a266fb34fd49b66e7c98b24d416bcdeae Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 28 Jul 2026 13:47:50 +0200 Subject: [PATCH] feat: add dataflow publication contracts and workflow webui --- src/govoplan_core/core/dataflows.py | 128 +++++++++++++++++++++ src/govoplan_core/core/datasources.py | 47 ++++++++ tests/test_dataflow_contract.py | 76 ++++++++++++ tests/test_datasource_contract.py | 22 ++++ webui/package-lock.json | 25 +++- webui/package.json | 3 +- webui/scripts/test-module-permutations.mjs | 7 +- webui/src/platform/modules.ts | 5 +- webui/vite.config.ts | 6 +- 9 files changed, 311 insertions(+), 8 deletions(-) create mode 100644 src/govoplan_core/core/dataflows.py create mode 100644 tests/test_dataflow_contract.py diff --git a/src/govoplan_core/core/dataflows.py b/src/govoplan_core/core/dataflows.py new file mode 100644 index 0000000..4b3f992 --- /dev/null +++ b/src/govoplan_core/core/dataflows.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from datetime import datetime +from typing import Protocol, runtime_checkable + + +CAPABILITY_DATAFLOW_RUN_LIFECYCLE = "dataflow.runLifecycle" + + +class DataflowRunError(ValueError): + """Stable base error for module-neutral Dataflow run operations.""" + + +class DataflowRunNotFoundError(DataflowRunError): + pass + + +class DataflowRunConflictError(DataflowRunError): + pass + + +class DataflowRunUnavailableError(DataflowRunError): + pass + + +@dataclass(frozen=True, slots=True) +class DataflowPublicationTarget: + target_datasource_ref: str | None = None + name: str | None = None + source_name: str | None = None + description: str | None = None + freeze: bool = False + frozen_label: str | None = None + set_current: bool = True + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class DataflowRunRequest: + pipeline_ref: str + revision: int + idempotency_key: str + row_limit: int = 500 + publication: DataflowPublicationTarget | None = None + + +@dataclass(frozen=True, slots=True) +class DataflowRunDescriptor: + ref: str + pipeline_ref: str + revision: int + status: str + definition_hash: str + executor_version: str + input_row_count: int = 0 + output_row_count: int = 0 + output_publication_ref: str | None = None + output_datasource_ref: str | None = None + output_materialization_ref: str | None = None + error: str | None = None + started_at: datetime | None = None + finished_at: datetime | None = None + replayed: bool = False + metadata: Mapping[str, object] = field(default_factory=dict) + + +@runtime_checkable +class DataflowRunLifecycleProvider(Protocol): + def start_run( + self, + session: object, + principal: object, + *, + request: DataflowRunRequest, + ) -> DataflowRunDescriptor: + ... + + def get_run( + self, + session: object, + principal: object, + *, + run_ref: str, + ) -> DataflowRunDescriptor | None: + ... + + def cancel_run( + self, + session: object, + principal: object, + *, + run_ref: str, + ) -> DataflowRunDescriptor: + ... + + +def dataflow_run_lifecycle( + registry: object | None, +) -> DataflowRunLifecycleProvider | None: + capability = _capability(registry, CAPABILITY_DATAFLOW_RUN_LIFECYCLE) + return capability if isinstance(capability, DataflowRunLifecycleProvider) else None + + +def _capability(registry: object | None, name: str) -> object | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not hasattr(registry, "capability") + or not registry.has_capability(name) + ): + return None + return registry.capability(name) + + +__all__ = [ + "CAPABILITY_DATAFLOW_RUN_LIFECYCLE", + "DataflowPublicationTarget", + "DataflowRunConflictError", + "DataflowRunDescriptor", + "DataflowRunError", + "DataflowRunLifecycleProvider", + "DataflowRunNotFoundError", + "DataflowRunRequest", + "DataflowRunUnavailableError", + "dataflow_run_lifecycle", +] diff --git a/src/govoplan_core/core/datasources.py b/src/govoplan_core/core/datasources.py index b2017f5..f1f75ad 100644 --- a/src/govoplan_core/core/datasources.py +++ b/src/govoplan_core/core/datasources.py @@ -8,6 +8,7 @@ from typing import Literal, Protocol, runtime_checkable CAPABILITY_DATASOURCE_CATALOGUE = "datasources.catalogue" CAPABILITY_DATASOURCE_LIFECYCLE = "datasources.lifecycle" +CAPABILITY_DATASOURCE_PUBLICATION = "datasources.publication" CAPABILITY_DATASOURCE_ORIGINS = "connectors.datasourceOrigins" DatasourceMode = Literal["live", "cached", "static"] @@ -152,6 +153,33 @@ class DatasourceStageInput: metadata: Mapping[str, object] = field(default_factory=dict) +@dataclass(frozen=True, slots=True) +class DatasourcePublicationRequest: + producer_module: str + producer_run_ref: str + idempotency_key: str + rows: tuple[Mapping[str, object], ...] + target_datasource_ref: str | None = None + name: str | None = None + source_name: str | None = None + description: str | None = None + freeze: bool = False + frozen_label: str | None = None + set_current: bool = True + source_timestamp: datetime | None = None + provenance: Mapping[str, object] = field(default_factory=dict) + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class DatasourcePublicationResult: + ref: str + status: str + datasource: DatasourceDescriptor + materialization: DatasourceMaterialization + replayed: bool = False + + @dataclass(frozen=True, slots=True) class DatasourceOrigin: ref: str @@ -302,6 +330,18 @@ class DatasourceLifecycleProvider(Protocol): ... +@runtime_checkable +class DatasourcePublicationProvider(Protocol): + def publish_rows( + self, + session: object, + principal: object, + *, + request: DatasourcePublicationRequest, + ) -> DatasourcePublicationResult: + ... + + @runtime_checkable class DatasourceOriginProvider(Protocol): def list_origins( @@ -343,6 +383,13 @@ def datasource_lifecycle(registry: object | None) -> DatasourceLifecycleProvider return capability if isinstance(capability, DatasourceLifecycleProvider) else None +def datasource_publication( + registry: object | None, +) -> DatasourcePublicationProvider | None: + capability = _capability(registry, CAPABILITY_DATASOURCE_PUBLICATION) + return capability if isinstance(capability, DatasourcePublicationProvider) else None + + def datasource_origins(registry: object | None) -> DatasourceOriginProvider | None: capability = _capability(registry, CAPABILITY_DATASOURCE_ORIGINS) return capability if isinstance(capability, DatasourceOriginProvider) else None diff --git a/tests/test_dataflow_contract.py b/tests/test_dataflow_contract.py new file mode 100644 index 0000000..d63d4ff --- /dev/null +++ b/tests/test_dataflow_contract.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.dataflows import ( + CAPABILITY_DATAFLOW_RUN_LIFECYCLE, + DataflowRunDescriptor, + DataflowRunLifecycleProvider, + dataflow_run_lifecycle, +) +from govoplan_core.core.modules import ModuleContext, ModuleManifest +from govoplan_core.core.registry import PlatformRegistry + + +class _Provider: + def start_run(self, session, principal, *, request): + del session, principal + return DataflowRunDescriptor( + ref="dataflow-run:1", + pipeline_ref=request.pipeline_ref, + revision=request.revision, + status="succeeded", + definition_hash="abc", + executor_version="test", + ) + + def get_run(self, session, principal, *, run_ref): + del session, principal + if run_ref != "dataflow-run:1": + return None + return DataflowRunDescriptor( + ref=run_ref, + pipeline_ref="pipeline:1", + revision=1, + status="succeeded", + definition_hash="abc", + executor_version="test", + ) + + def cancel_run(self, session, principal, *, run_ref): + del session, principal + return DataflowRunDescriptor( + ref=run_ref, + pipeline_ref="pipeline:1", + revision=1, + status="cancelled", + definition_hash="abc", + executor_version="test", + ) + + +class DataflowContractTests(unittest.TestCase): + def test_run_lifecycle_is_runtime_checkable_and_resolved(self) -> None: + provider = _Provider() + self.assertIsInstance(provider, DataflowRunLifecycleProvider) + registry = PlatformRegistry() + registry.register( + ModuleManifest( + id="dataflow_contract_test", + name="Dataflow contract test", + version="test", + capability_factories={ + CAPABILITY_DATAFLOW_RUN_LIFECYCLE: lambda context: provider, + }, + ) + ) + registry.configure_capability_context( + ModuleContext(registry=registry, settings=object()) + ) + + self.assertIs(provider, dataflow_run_lifecycle(registry)) + self.assertIsNone(dataflow_run_lifecycle(PlatformRegistry())) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_datasource_contract.py b/tests/test_datasource_contract.py index c0bb7a7..05c92bc 100644 --- a/tests/test_datasource_contract.py +++ b/tests/test_datasource_contract.py @@ -6,6 +6,7 @@ from govoplan_core.core.datasources import ( CAPABILITY_DATASOURCE_CATALOGUE, CAPABILITY_DATASOURCE_LIFECYCLE, CAPABILITY_DATASOURCE_ORIGINS, + CAPABILITY_DATASOURCE_PUBLICATION, DatasourceCatalogueProvider, DatasourceDescriptor, DatasourceField, @@ -14,11 +15,14 @@ from govoplan_core.core.datasources import ( DatasourceOrigin, DatasourceOriginProvider, DatasourceOriginReadResult, + DatasourcePublicationProvider, + DatasourcePublicationResult, DatasourceReadResult, DatasourceStage, datasource_catalogue, datasource_lifecycle, datasource_origins, + datasource_publication, ) from govoplan_core.core.modules import ModuleContext, ModuleManifest from govoplan_core.core.registry import PlatformRegistry @@ -99,6 +103,21 @@ class _Provider: del session, principal, datasource_ref return self.descriptor + def publish_rows(self, session, principal, *, request): + del session, principal, request + return DatasourcePublicationResult( + ref="publication:1", + status="published", + datasource=self.descriptor, + materialization=DatasourceMaterialization( + ref="materialization:1", + datasource_ref=self.descriptor.ref, + revision=1, + state="published", + fingerprint=self.descriptor.fingerprint, + ), + ) + def list_origins(self, session, principal, *, query="", limit=100): del session, principal, query return ( @@ -131,6 +150,7 @@ class DatasourceContractTests(unittest.TestCase): provider = _Provider() self.assertIsInstance(provider, DatasourceCatalogueProvider) self.assertIsInstance(provider, DatasourceLifecycleProvider) + self.assertIsInstance(provider, DatasourcePublicationProvider) self.assertIsInstance(provider, DatasourceOriginProvider) registry = PlatformRegistry() registry.register( @@ -141,6 +161,7 @@ class DatasourceContractTests(unittest.TestCase): capability_factories={ CAPABILITY_DATASOURCE_CATALOGUE: lambda context: provider, CAPABILITY_DATASOURCE_LIFECYCLE: lambda context: provider, + CAPABILITY_DATASOURCE_PUBLICATION: lambda context: provider, CAPABILITY_DATASOURCE_ORIGINS: lambda context: provider, }, ) @@ -149,6 +170,7 @@ class DatasourceContractTests(unittest.TestCase): self.assertIs(provider, datasource_catalogue(registry)) self.assertIs(provider, datasource_lifecycle(registry)) + self.assertIs(provider, datasource_publication(registry)) self.assertIs(provider, datasource_origins(registry)) self.assertIsNone(datasource_catalogue(PlatformRegistry())) diff --git a/webui/package-lock.json b/webui/package-lock.json index 210d719..88e8f4f 100644 --- a/webui/package-lock.json +++ b/webui/package-lock.json @@ -25,7 +25,8 @@ "@govoplan/ops-webui": "file:../../govoplan-ops/webui", "@govoplan/organizations-webui": "file:../../govoplan-organizations/webui", "@govoplan/policy-webui": "file:../../govoplan-policy/webui", - "@govoplan/scheduling-webui": "file:../../govoplan-scheduling/webui" + "@govoplan/scheduling-webui": "file:../../govoplan-scheduling/webui", + "@govoplan/workflow-webui": "file:../../govoplan-workflow/webui" }, "devDependencies": { "@types/react": "^19.0.2", @@ -377,6 +378,24 @@ } } }, + "../../govoplan-workflow/webui": { + "name": "@govoplan/workflow-webui", + "version": "0.1.14", + "peerDependencies": { + "@govoplan/core-webui": "^0.1.14", + "@xyflow/react": "^12.11.2", + "lucide-react": "^1.23.0", + "react": "^19.0.0", + "react-dom": "^19.0.0", + "react-router-dom": "^7.1.1", + "typescript": "^5.7.2" + }, + "peerDependenciesMeta": { + "@govoplan/core-webui": { + "optional": true + } + } + }, "node_modules/@babel/code-frame": { "version": "7.29.7", "resolved": "https://registry.npmjs.org/@babel/code-frame/-/code-frame-7.29.7.tgz", @@ -1173,6 +1192,10 @@ "resolved": "../../govoplan-scheduling/webui", "link": true }, + "node_modules/@govoplan/workflow-webui": { + "resolved": "../../govoplan-workflow/webui", + "link": true + }, "node_modules/@jridgewell/gen-mapping": { "version": "0.3.13", "resolved": "https://registry.npmjs.org/@jridgewell/gen-mapping/-/gen-mapping-0.3.13.tgz", diff --git a/webui/package.json b/webui/package.json index 50553b4..2ce3e56 100644 --- a/webui/package.json +++ b/webui/package.json @@ -56,7 +56,8 @@ "@govoplan/organizations-webui": "file:../../govoplan-organizations/webui", "@govoplan/ops-webui": "file:../../govoplan-ops/webui", "@govoplan/policy-webui": "file:../../govoplan-policy/webui", - "@govoplan/scheduling-webui": "file:../../govoplan-scheduling/webui" + "@govoplan/scheduling-webui": "file:../../govoplan-scheduling/webui", + "@govoplan/workflow-webui": "file:../../govoplan-workflow/webui" }, "devDependencies": { "@types/react": "^19.0.2", diff --git a/webui/scripts/test-module-permutations.mjs b/webui/scripts/test-module-permutations.mjs index d1c6b53..edbe0f3 100644 --- a/webui/scripts/test-module-permutations.mjs +++ b/webui/scripts/test-module-permutations.mjs @@ -18,7 +18,8 @@ const packageByModule = { organizations: "@govoplan/organizations-webui", ops: "@govoplan/ops-webui", policy: "@govoplan/policy-webui", - scheduling: "@govoplan/scheduling-webui" + scheduling: "@govoplan/scheduling-webui", + workflow: "@govoplan/workflow-webui" }; const cases = [ @@ -32,6 +33,8 @@ const cases = [ { name: "dataflow-only", modules: ["dataflow"] }, { name: "datasources-only", modules: ["datasources"] }, { name: "dataflow-with-datasources", modules: ["datasources", "dataflow"] }, + { name: "workflow-only", modules: ["workflow"] }, + { name: "workflow-with-dataflow", modules: ["datasources", "dataflow", "workflow"] }, { name: "calendar-only", modules: ["calendar"] }, { name: "files-only", modules: ["files"] }, { name: "mail-only", modules: ["mail"] }, @@ -44,7 +47,7 @@ const cases = [ { name: "scheduling-only", modules: ["scheduling"] }, { name: "scheduling-with-calendar", modules: ["scheduling", "calendar"] }, { name: "docs-and-ops", modules: ["access", "docs", "ops"] }, - { name: "full-product", modules: ["access", "admin", "addresses", "policy", "audit", "dashboard", "datasources", "dataflow", "organizations", "idm", "campaigns", "files", "mail", "notifications", "docs", "ops", "calendar", "scheduling"] } + { name: "full-product", modules: ["access", "admin", "addresses", "policy", "audit", "dashboard", "datasources", "dataflow", "workflow", "organizations", "idm", "campaigns", "files", "mail", "notifications", "docs", "ops", "calendar", "scheduling"] } ]; const npmExec = process.env.npm_execpath; diff --git a/webui/src/platform/modules.ts b/webui/src/platform/modules.ts index b088fcb..f674b5a 100644 --- a/webui/src/platform/modules.ts +++ b/webui/src/platform/modules.ts @@ -1,4 +1,4 @@ -import { Activity, Bell, BookUser, Building2, CalendarClock, CalendarDays, ClipboardPenLine, DatabaseZap, Folder, Form, LayoutDashboard, LayoutTemplate, Mail, Mails, RadioTower, Shield, Users, Waypoints, type LucideIcon } from "lucide-react"; +import { Activity, Bell, BookUser, Building2, CalendarClock, CalendarDays, ClipboardPenLine, DatabaseZap, Folder, Form, LayoutDashboard, LayoutTemplate, Mail, Mails, RadioTower, Shield, Users, Waypoints, Workflow as WorkflowIcon, type LucideIcon } from "lucide-react"; import installedWebModules from "virtual:govoplan-installed-modules"; import type { AuthInfo, DashboardWidgetContribution, DashboardWidgetsUiCapability, PlatformModuleInfo, PlatformNavItem, PlatformPublicModuleInfo, PlatformWebModule } from "../types"; import { @@ -63,7 +63,8 @@ const iconByName: Record = { reports: ClipboardPenLine, templates: LayoutTemplate, users: Users, - waypoints: Waypoints + waypoints: Waypoints, + workflow: WorkflowIcon }; export function iconComponentForName(iconName: string | null | undefined): LucideIcon | undefined { diff --git a/webui/vite.config.ts b/webui/vite.config.ts index cc4e5ed..f896497 100644 --- a/webui/vite.config.ts +++ b/webui/vite.config.ts @@ -28,7 +28,8 @@ const defaultWebModulePackages = [ "@govoplan/organizations-webui", "@govoplan/ops-webui", "@govoplan/policy-webui", - "@govoplan/scheduling-webui" + "@govoplan/scheduling-webui", + "@govoplan/workflow-webui" ]; function configuredWebModulePackages(): string[] { @@ -115,7 +116,8 @@ export default defineConfig({ fileURLToPath(new URL('../../govoplan-campaign/webui', import.meta.url)), fileURLToPath(new URL('../../govoplan-ops/webui', import.meta.url)), fileURLToPath(new URL('../../govoplan-policy/webui', import.meta.url)), - fileURLToPath(new URL('../../govoplan-scheduling/webui', import.meta.url)) + fileURLToPath(new URL('../../govoplan-scheduling/webui', import.meta.url)), + fileURLToPath(new URL('../../govoplan-workflow/webui', import.meta.url)) ] }, proxy: {