GovOPlaN/govoplan-dataflow
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: consume governed datasources and shared graphs

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-28 12:44:46 +02:00
parent dee8380631
commit 6ca3058021
15 changed files with 248 additions and 272 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/govoplan_dataflow/backend/executor.py
View File

@@ -137,7 +137,7 @@ def execute_preview(
elif node.type == "source.reference":
if source_resolver is None:
raise PipelineExecutionError(
"Connector-backed preview requires the Connectors tabular-source capability.",
"Datasource-backed preview requires the Datasources catalogue capability.",
node_id=node.id,
)
resolved = source_resolver(node, MAX_SOURCE_ROWS)
@@ -148,7 +148,7 @@ def execute_preview(
"node_id": node.id,
"source_ref": resolved.source_ref,
"source_name": node.config.get("source_name"),
"kind": "connector",
"kind": "datasource",
"provider": resolved.provider,
"fingerprint": resolved.fingerprint,
"row_count": resolved.total_rows,

158
src/govoplan_dataflow/backend/graph.py
View File

@@ -7,7 +7,16 @@ from collections import deque
from dataclasses import dataclass
from typing import Any
from govoplan_dataflow.backend.node_library import NODE_TYPES, node_definition
from govoplan_core.core.definition_graphs import (
DefinitionEdge,
DefinitionNode,
validate_definition_graph,
)
from govoplan_dataflow.backend.node_library import (
DATAFLOW_GRAPH_LIBRARY,
NODE_TYPES,
node_definition,
)
from govoplan_dataflow.backend.schemas import DataflowDiagnostic, GraphEdge, GraphNode, PipelineGraph
@@ -47,75 +56,49 @@ def definition_hash(graph: PipelineGraph, sql_text: str | None = None) -> str:
def validate_graph(graph: PipelineGraph) -> list[DataflowDiagnostic]:
diagnostics: list[DataflowDiagnostic] = []
diagnostics = [
DataflowDiagnostic(
severity=item.severity,
code=item.code,
message=item.message,
node_id=item.node_id,
field=item.field,
)
for item in validate_definition_graph(
DATAFLOW_GRAPH_LIBRARY,
nodes=tuple(DefinitionNode(id=node.id, type=node.type) for node in graph.nodes),
edges=tuple(
DefinitionEdge(
id=edge.id,
source=edge.source,
target=edge.target,
source_port=edge.source_port,
target_port=edge.target_port,
)
for edge in graph.edges
),
)
]
nodes = {node.id: node for node in graph.nodes}
if len(nodes) != len(graph.nodes):
diagnostics.append(_error("graph.duplicate_node", "Node identifiers must be unique."))
edge_ids = {edge.id for edge in graph.edges}
if len(edge_ids) != len(graph.edges):
diagnostics.append(_error("graph.duplicate_edge", "Edge identifiers must be unique."))
incoming: dict[str, list[GraphEdge]] = {node_id: [] for node_id in nodes}
outgoing: dict[str, list[str]] = {node_id: [] for node_id in nodes}
for edge in graph.edges:
if edge.source not in nodes:
diagnostics.append(
_error("edge.unknown_source", f"Edge {edge.id!r} references an unknown source node.")
)
continue
if edge.target not in nodes:
diagnostics.append(
_error("edge.unknown_target", f"Edge {edge.id!r} references an unknown target node.")
)
continue
if edge.source == edge.target:
diagnostics.append(
_error("edge.self_reference", "A node cannot connect to itself.", node_id=edge.source)
)
if edge.source not in nodes or edge.target not in nodes or edge.source == edge.target:
continue
source_definition = node_definition(nodes[edge.source].type)
target_definition = node_definition(nodes[edge.target].type)
if source_definition and edge.source_port not in {
port.id for port in source_definition.output_ports
}:
diagnostics.append(
_error(
"edge.unknown_source_port",
f"Node {edge.source!r} has no output port {edge.source_port!r}.",
node_id=edge.source,
)
)
continue
if target_definition and edge.target_port not in {
port.id for port in target_definition.input_ports
}:
diagnostics.append(
_error(
"edge.unknown_target_port",
f"Node {edge.target!r} has no input port {edge.target_port!r}.",
node_id=edge.target,
)
)
if (
source_definition is None
or target_definition is None
or edge.source_port not in {port.id for port in source_definition.output_ports}
or edge.target_port not in {port.id for port in target_definition.input_ports}
):
continue
outgoing[edge.source].append(edge.target)
incoming[edge.target].append(edge)
if not graph.nodes:
diagnostics.append(_error("graph.empty", "Add a source and an output before saving the pipeline."))
return diagnostics
source_nodes = [node for node in graph.nodes if node.type.startswith("source.")]
output_nodes = [node for node in graph.nodes if node.type == "output"]
if not source_nodes:
diagnostics.append(
_error(
"graph.source_count",
"A pipeline needs at least one source node.",
)
)
elif len(source_nodes) > 10:
diagnostics.append(_error("graph.source_limit", "Pipelines are limited to ten sources."))
source_names = [
str(node.config.get("source_name", "")).strip()
for node in source_nodes
@@ -137,54 +120,13 @@ def validate_graph(graph: PipelineGraph) -> list[DataflowDiagnostic]:
f"{', '.join(duplicate_source_names)}.",
)
)
if len(output_nodes) != 1:
diagnostics.append(
_error("graph.output_count", "A pipeline must contain exactly one output node.")
)
for node in graph.nodes:
if node.type not in SUPPORTED_NODE_TYPES:
diagnostics.append(
_error(
"node.unsupported_type",
f"Node type {node.type!r} is not supported by this executor.",
node_id=node.id,
field="type",
)
)
continue
definition = node_definition(node.type)
node_edges = incoming.get(node.id, [])
if definition is not None:
for port in definition.input_ports:
connections = [
edge
for edge in node_edges
if edge.target_port == port.id
]
minimum = port.minimum_connections if port.required else 0
if len(connections) < minimum:
diagnostics.append(
_error(
"node.input_required",
f"{definition.label} requires {port.label.lower()} input.",
node_id=node.id,
)
)
if not port.multiple and len(connections) > 1:
diagnostics.append(
_error(
"node.input_multiple",
f"{port.label} accepts only one connection.",
node_id=node.id,
)
)
diagnostics.extend(_validate_node_config(node))
ordered, cyclic = topological_order(graph)
if cyclic:
diagnostics.append(_error("graph.cycle", "Pipeline edges must form an acyclic graph."))
elif source_nodes and output_nodes:
if not cyclic and source_nodes and len(output_nodes) == 1:
reachable: set[str] = set()
for source in source_nodes:
reachable.update(_reachable_from(source.id, outgoing))
@@ -206,7 +148,21 @@ def validate_graph(graph: PipelineGraph) -> list[DataflowDiagnostic]:
_error("graph.output_not_terminal", "The output node must be the terminal transform.")
)
diagnostics.extend(_validate_graph_schemas(graph, ordered=ordered))
return diagnostics
return _dedupe_diagnostics(diagnostics)
def _dedupe_diagnostics(
diagnostics: list[DataflowDiagnostic],
) -> list[DataflowDiagnostic]:
seen: set[tuple[str, str | None, str | None]] = set()
result: list[DataflowDiagnostic] = []
for item in diagnostics:
key = (item.code, item.node_id, item.field)
if key in seen:
continue
seen.add(key)
result.append(item)
return result
def topological_order(graph: PipelineGraph) -> tuple[list[str], bool]:
@@ -577,7 +533,7 @@ def _validate_node_config(node: GraphNode) -> list[DataflowDiagnostic]:
diagnostics.append(
_error(
"source.reference_required",
"Choose a connector source.",
"Choose a datasource.",
node_id=node.id,
field="source_ref",
)

24
src/govoplan_dataflow/backend/manifest.py
View File

@@ -18,9 +18,9 @@ from govoplan_core.core.modules import (
PermissionDefinition,
RoleTemplate,
)
from govoplan_core.core.tabular_sources import (
CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER,
CAPABILITY_CONNECTORS_TABULAR_SOURCES,
from govoplan_core.core.datasources import (
CAPABILITY_DATASOURCE_CATALOGUE,
CAPABILITY_DATASOURCE_LIFECYCLE,
)
from govoplan_core.db.base import Base
from govoplan_dataflow.backend.db import models as dataflow_models
@@ -101,8 +101,9 @@ DOCUMENTATION = (
summary="Versioned tabular transformations with graphical and constrained SQL editing.",
body=(
"Dataflow owns canonical pipeline graphs, immutable revisions, validation, constrained "
"SQL compilation, preview and run diagnostics, and lineage references. Connectors owns "
"source connections and credentials; Reporting owns analytical presentation and exports; "
"SQL compilation, preview and run diagnostics, and lineage references. Datasources owns "
"the governed catalogue and materializations, while Connectors owns external acquisition "
"and credentials; Reporting owns analytical presentation and exports; "
"Workflow owns orchestration and human handoffs; Risk Compliance owns sanctions review "
"semantics and policy gates. User SQL is compiled into approved transforms and is never "
"passed unchecked to a backing database."
@@ -112,6 +113,7 @@ DOCUMENTATION = (
audience=("operator", "module_admin", "power_user", "product_owner"),
order=75,
related_modules=(
"datasources",
"connectors",
"files",
"reporting",
@@ -123,7 +125,7 @@ DOCUMENTATION = (
),
metadata={
"first_slice": (
"Inline and connector sources, union, join, filter, deduplication, select, "
"Inline and governed datasources, union, join, filter, deduplication, select, "
"derived columns, aggregate, sort, limit, output, revisioning, and bounded preview."
),
"sql_safety": "Constrained AST compilation only; no pass-through execution.",
@@ -167,7 +169,7 @@ manifest = ModuleManifest(
optional_dependencies=(
"access",
"audit",
"connectors",
"datasources",
"files",
"notifications",
"policy",
@@ -176,8 +178,8 @@ manifest = ModuleManifest(
"workflow",
),
optional_capabilities=(
CAPABILITY_CONNECTORS_TABULAR_SOURCES,
CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER,
CAPABILITY_DATASOURCE_CATALOGUE,
CAPABILITY_DATASOURCE_LIFECYCLE,
),
provides_interfaces=(
ModuleInterfaceProvider(name="dataflow.pipeline_catalog", version=MODULE_VERSION),
@@ -187,13 +189,13 @@ manifest = ModuleManifest(
),
requires_interfaces=(
ModuleInterfaceRequirement(
name="connectors.tabular_sources",
name="datasources.catalogue",
version_min="0.1.0",
version_max_exclusive="1.0.0",
optional=True,
),
ModuleInterfaceRequirement(
name="connectors.tabular_snapshot_writer",
name="datasources.lifecycle",
version_min="0.1.0",
version_max_exclusive="1.0.0",
optional=True,

100
src/govoplan_dataflow/backend/node_library.py
View File

@@ -1,49 +1,30 @@
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Literal
from dataclasses import dataclass
from typing import Literal
from govoplan_core.core.definition_graphs import (
DefinitionConfigField,
DefinitionGraphConstraints,
DefinitionGraphLibrary,
DefinitionNodeCountConstraint,
DefinitionNodeType,
DefinitionPort,
)
NodeCategory = Literal["load", "combine", "filter", "transform", "output"]
SqlSupport = Literal["full", "partial", "none"]
NodePortDefinition = DefinitionPort
NodeConfigField = DefinitionConfigField
@dataclass(frozen=True, slots=True)
class NodePortDefinition:
id: str
label: str
required: bool = True
multiple: bool = False
minimum_connections: int = 1
@dataclass(frozen=True, slots=True)
class NodeConfigField:
id: str
label: str
kind: str
required: bool = False
description: str | None = None
options: tuple[tuple[str, str], ...] = ()
@dataclass(frozen=True, slots=True)
class NodeTypeDefinition:
type: str
category: NodeCategory
label: str
description: str
icon: str
input_ports: tuple[NodePortDefinition, ...] = ()
output_ports: tuple[NodePortDefinition, ...] = (
NodePortDefinition(id="output", label="Output"),
)
config_fields: tuple[NodeConfigField, ...] = ()
default_config: dict[str, Any] = field(default_factory=dict)
class NodeTypeDefinition(DefinitionNodeType):
sql_support: SqlSupport = "full"
NODE_LIBRARY = (
_NODE_TYPES = (
NodeTypeDefinition(
type="source.inline",
category="load",
@@ -59,18 +40,29 @@ NODE_LIBRARY = (
NodeTypeDefinition(
type="source.reference",
category="load",
label="Connector source",
description="Load a bounded, fingerprinted table exposed by Connectors.",
label="Datasource",
description="Load a bounded, fingerprinted state from the Datasources catalogue.",
icon="database",
config_fields=(
NodeConfigField(id="source_ref", label="Source", kind="tabular_source", required=True),
NodeConfigField(id="source_ref", label="Datasource", kind="datasource", required=True),
NodeConfigField(id="source_name", label="SQL source name", kind="text", required=True),
NodeConfigField(id="expected_fingerprint", label="Expected fingerprint", kind="readonly"),
NodeConfigField(
id="consistency",
label="State",
kind="select",
options=(
("current", "Current"),
("live", "Live"),
("frozen", "Latest frozen"),
),
),
),
default_config={
"source_ref": "",
"source_name": "connector_source",
"source_name": "datasource",
"expected_fingerprint": "",
"consistency": "current",
},
),
NodeTypeDefinition(
@@ -278,7 +270,6 @@ NODE_LIBRARY = (
),
)
NODE_TYPES = {definition.type: definition for definition in NODE_LIBRARY}
CATEGORY_LABELS = {
"load": "Load",
"combine": "Combine",
@@ -286,6 +277,36 @@ CATEGORY_LABELS = {
"transform": "Transform",
"output": "Output",
}
DATAFLOW_GRAPH_LIBRARY = DefinitionGraphLibrary(
id="dataflow",
version="0.1.0",
category_labels=CATEGORY_LABELS,
node_types=_NODE_TYPES,
constraints=DefinitionGraphConstraints(
max_nodes=100,
max_edges=200,
allow_cycles=False,
require_connected=True,
node_counts=(
DefinitionNodeCountConstraint(
code="graph.source_count",
label="source",
minimum=1,
maximum=10,
type_prefixes=("source.",),
),
DefinitionNodeCountConstraint(
code="graph.output_count",
label="output",
minimum=1,
maximum=1,
node_types=("output",),
),
),
),
)
NODE_LIBRARY: tuple[NodeTypeDefinition, ...] = _NODE_TYPES
NODE_TYPES = {definition.type: definition for definition in NODE_LIBRARY}
def node_definition(node_type: str) -> NodeTypeDefinition | None:
@@ -294,6 +315,7 @@ def node_definition(node_type: str) -> NodeTypeDefinition | None:
__all__ = [
"CATEGORY_LABELS",
"DATAFLOW_GRAPH_LIBRARY",
"NODE_LIBRARY",
"NODE_TYPES",
"NodeCategory",

71
src/govoplan_dataflow/backend/router.py
View File

@@ -5,16 +5,19 @@ from sqlalchemy.orm import Session
from govoplan_core.audit.logging import audit_event
from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
from govoplan_core.core.datasources import (
DatasourceAccessError,
DatasourceDescriptor,
DatasourceError,
DatasourceNotFoundError,
DatasourceStageInput,
DatasourceUnavailableError,
DatasourceValidationError,
datasource_catalogue,
datasource_lifecycle,
)
from govoplan_core.core.tabular_sources import (
TabularSnapshotInput,
TabularSource,
TabularSourceAccessError,
TabularSourceError,
TabularSourceNotFoundError,
TabularSourceValidationError,
parse_tabular_csv,
tabular_snapshot_writer,
tabular_source_provider,
)
from govoplan_core.db.session import get_session
from govoplan_dataflow.backend.manifest import ADMIN_SCOPE, READ_SCOPE, RUN_SCOPE, WRITE_SCOPE
@@ -93,12 +96,14 @@ def _http_error(exc: DataflowError) -> HTTPException:
return HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))
def _source_http_error(exc: TabularSourceError) -> HTTPException:
if isinstance(exc, TabularSourceAccessError):
def _source_http_error(exc: DatasourceError) -> HTTPException:
if isinstance(exc, DatasourceAccessError):
return HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
if isinstance(exc, TabularSourceNotFoundError):
if isinstance(exc, DatasourceNotFoundError):
return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
if isinstance(exc, TabularSourceValidationError):
if isinstance(exc, DatasourceUnavailableError):
return HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc))
if isinstance(exc, DatasourceValidationError):
return HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail=str(exc),
@@ -155,10 +160,10 @@ def _node_library_response() -> NodeLibraryResponse:
)
def _source_response(source: TabularSource) -> TabularSourceResponse:
def _source_response(source: DatasourceDescriptor) -> TabularSourceResponse:
return TabularSourceResponse(
ref=source.ref,
provider=source.provider,
provider=source.provider or "datasources",
source_name=source.source_name,
name=source.name,
description=source.description,
@@ -195,18 +200,18 @@ def api_list_sources(
) -> TabularSourceListResponse:
_require_any_scope(principal, READ_SCOPE, WRITE_SCOPE, RUN_SCOPE, ADMIN_SCOPE)
registry = get_registry()
provider = tabular_source_provider(registry)
writer = tabular_snapshot_writer(registry)
provider = datasource_catalogue(registry)
writer = datasource_lifecycle(registry)
if provider is None:
return TabularSourceListResponse(available=False, writable=False, sources=[])
try:
sources = provider.list_sources(
sources = provider.list_datasources(
session,
principal,
query=query,
limit=100,
)
except TabularSourceError as exc:
except DatasourceError as exc:
raise _source_http_error(exc) from exc
return TabularSourceListResponse(
available=True,
@@ -226,11 +231,11 @@ def api_create_source_snapshot(
principal: ApiPrincipal = Depends(get_api_principal),
) -> TabularSourceResponse:
_require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
writer = tabular_snapshot_writer(get_registry())
writer = datasource_lifecycle(get_registry())
if writer is None:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="No tabular snapshot writer is available.",
detail="No datasource staging provider is available.",
)
try:
rows = (
@@ -242,35 +247,49 @@ def api_create_source_snapshot(
if payload.format == "csv"
else tuple(payload.rows or ())
)
source = writer.create_snapshot(
stage = writer.create_stage(
session,
principal,
snapshot=TabularSnapshotInput(
stage=DatasourceStageInput(
name=payload.name,
source_name=payload.source_name,
description=payload.description,
kind="upload",
mode="static",
shape="tabular",
rows=rows,
metadata={
provider="dataflow.upload",
provenance={
"created_via": "dataflow",
"source_format": payload.format,
},
metadata={
"dataflow_convenience_import": True,
},
),
)
except TabularSourceError as exc:
source, materialization = writer.promote_stage(
session,
principal,
stage_ref=stage.ref,
)
except DatasourceError as exc:
raise _source_http_error(exc) from exc
audit_event(
session,
tenant_id=principal.tenant_id,
user_id=getattr(principal.user, "id", None),
api_key_id=principal.api_key_id,
action="dataflow.source_snapshot.created",
object_type="tabular_source",
action="dataflow.datasource.created",
object_type="datasource",
object_id=source.ref,
details={
"provider": source.provider,
"source_name": source.source_name,
"row_count": source.row_count,
"fingerprint": source.fingerprint,
"stage_ref": stage.ref,
"materialization_ref": materialization.ref,
},
)
response = _source_response(source)

31
src/govoplan_dataflow/backend/service.py
View File

@@ -6,10 +6,10 @@ from sqlalchemy import select
from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.tabular_sources import (
TabularReadRequest,
TabularSourceError,
tabular_source_provider,
from govoplan_core.core.datasources import (
DatasourceError,
DatasourceReadRequest,
datasource_catalogue,
)
from govoplan_core.db.base import utcnow
from govoplan_dataflow.backend.db.models import (
@@ -381,38 +381,41 @@ def preview_pipeline(
started_at = utcnow()
run: DataflowRun | None = None
try:
provider = tabular_source_provider(registry)
provider = datasource_catalogue(registry)
def resolve_source(node: GraphNode, limit: int) -> ResolvedSource:
if provider is None:
raise PipelineExecutionError(
"Connector-backed preview requires the Connectors tabular-source capability.",
"Datasource-backed preview requires the Datasources catalogue capability.",
node_id=node.id,
)
if principal is None:
raise PipelineExecutionError(
"Connector-backed preview requires a tenant API principal.",
"Datasource-backed preview requires a tenant API principal.",
node_id=node.id,
)
try:
resolved = provider.read_source(
resolved = provider.read_datasource(
session,
principal,
request=TabularReadRequest(
source_ref=str(node.config["source_ref"]),
request=DatasourceReadRequest(
datasource_ref=str(node.config["source_ref"]),
consistency=str(
node.config.get("consistency") or "current"
), # type: ignore[arg-type]
limit=limit,
expected_fingerprint=_clean_optional(
node.config.get("expected_fingerprint")
),
),
)
except TabularSourceError as exc:
except DatasourceError as exc:
raise PipelineExecutionError(str(exc), node_id=node.id) from exc
return ResolvedSource(
rows=tuple(dict(row) for row in resolved.rows),
source_ref=resolved.source.ref,
provider=resolved.source.provider,
fingerprint=resolved.source.fingerprint,
source_ref=resolved.datasource.ref,
provider=resolved.datasource.provider or "datasources",
fingerprint=resolved.datasource.fingerprint,
total_rows=resolved.total_rows,
truncated=resolved.truncated,
)