feat: add dataflow publication contracts and workflow webui

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",
+]

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