diff --git a/README.md b/README.md index ec91477..195e336 100644 --- a/README.md +++ b/README.md @@ -10,8 +10,10 @@ the previewed row contents. - **Dataflow:** pipeline graphs, revisions, validation, constrained SQL, previews/runs, diagnostics, and lineage. -- **Connectors:** source connections, credentials, discovery, health, and - bounded source access. +- **Datasources:** governed source identity, staging, materializations, frozen + states, and bounded source access. +- **Connectors:** external acquisition, credentials, discovery, and provider + health. - **Reporting:** governed datasets, analytical views, dashboards, and exports. - **Workflow:** orchestration, resumability, approvals, and module handoffs. - **Risk Compliance:** sanctions matching policy, review, dispositions, and @@ -24,7 +26,7 @@ nodes by purpose: | Group | Nodes | | --- | --- | -| Load | Inline data, connector source | +| Load | Inline data, datasource | | Combine | Append rows, join tables | | Filter | Filter rows, remove duplicates | | Transform | Select columns, derive column, aggregate, sort, limit | @@ -34,10 +36,11 @@ Join nodes have explicit left and right ports. Append nodes accept two or more inputs. Derived columns use a constrained operation catalogue rather than arbitrary code. -Connector sources are resolved through the versioned Core capability -`connectors.tabular_sources`; Dataflow does not import Connectors or store its -credentials. Source references pin an expected fingerprint so schema/content -drift fails visibly instead of silently changing a run. +Governed inputs are resolved through the versioned Core capability +`datasources.catalogue`; Dataflow imports neither Datasources nor Connectors and +stores only opaque datasource references plus expected fingerprints. A source +node can request the current, live, or latest frozen state. Fingerprint drift +fails visibly instead of silently changing a run. ## SQL And Preview Safety diff --git a/src/govoplan_dataflow/backend/executor.py b/src/govoplan_dataflow/backend/executor.py index a565a5f..45a7742 100644 --- a/src/govoplan_dataflow/backend/executor.py +++ b/src/govoplan_dataflow/backend/executor.py @@ -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, diff --git a/src/govoplan_dataflow/backend/graph.py b/src/govoplan_dataflow/backend/graph.py index c98fb16..e15ef8b 100644 --- a/src/govoplan_dataflow/backend/graph.py +++ b/src/govoplan_dataflow/backend/graph.py @@ -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", ) diff --git a/src/govoplan_dataflow/backend/manifest.py b/src/govoplan_dataflow/backend/manifest.py index 6c9bbe7..e0903ee 100644 --- a/src/govoplan_dataflow/backend/manifest.py +++ b/src/govoplan_dataflow/backend/manifest.py @@ -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, diff --git a/src/govoplan_dataflow/backend/node_library.py b/src/govoplan_dataflow/backend/node_library.py index bd04b39..c8fd852 100644 --- a/src/govoplan_dataflow/backend/node_library.py +++ b/src/govoplan_dataflow/backend/node_library.py @@ -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", diff --git a/src/govoplan_dataflow/backend/router.py b/src/govoplan_dataflow/backend/router.py index f2857b8..f3d65c0 100644 --- a/src/govoplan_dataflow/backend/router.py +++ b/src/govoplan_dataflow/backend/router.py @@ -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) diff --git a/src/govoplan_dataflow/backend/service.py b/src/govoplan_dataflow/backend/service.py index a2c4831..5d43b2b 100644 --- a/src/govoplan_dataflow/backend/service.py +++ b/src/govoplan_dataflow/backend/service.py @@ -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, ) diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 0b75f33..907a619 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -13,7 +13,7 @@ class DataflowManifestTests(unittest.TestCase): self.assertIsInstance(manifest, ModuleManifest) self.assertEqual(manifest.id, "dataflow") self.assertEqual(manifest.dependencies, ()) - self.assertIn("connectors", manifest.optional_dependencies) + self.assertIn("datasources", manifest.optional_dependencies) self.assertIn("reporting", manifest.optional_dependencies) self.assertIn("workflow", manifest.optional_dependencies) self.assertEqual(manifest.frontend.package_name, "@govoplan/dataflow-webui") diff --git a/webui/src/api/dataflow.ts b/webui/src/api/dataflow.ts index 27ea6d9..86a47b2 100644 --- a/webui/src/api/dataflow.ts +++ b/webui/src/api/dataflow.ts @@ -1,4 +1,5 @@ import { apiFetch, type ApiSettings } from "@govoplan/core-webui"; +import type { DefinitionGraphNodeType } from "@govoplan/core-webui/definition-graph"; export type PipelineStatus = "draft" | "active" | "archived"; export type EditorMode = "graph" | "sql"; @@ -125,17 +126,11 @@ export type NodeConfigFieldDefinition = { options: Array<[string, string]>; }; -export type NodeTypeDefinition = { - type: string; +export type NodeTypeDefinition = DefinitionGraphNodeType & { category: NodeCategory; - category_label: string; - label: string; - description: string; - icon: string; input_ports: NodePortDefinition[]; output_ports: NodePortDefinition[]; config_fields: NodeConfigFieldDefinition[]; - default_config: Record; sql_support: "full" | "partial" | "none"; }; diff --git a/webui/src/features/dataflow/DataflowCanvas.tsx b/webui/src/features/dataflow/DataflowCanvas.tsx index a2b8f07..87ff76d 100644 --- a/webui/src/features/dataflow/DataflowCanvas.tsx +++ b/webui/src/features/dataflow/DataflowCanvas.tsx @@ -13,6 +13,7 @@ import { type Edge, type ReactFlowInstance } from "@xyflow/react"; +import { definitionConnectionError } from "@govoplan/core-webui/definition-graph"; import type { DataflowDiagnostic, NodeTypeDefinition, @@ -124,34 +125,17 @@ export default function DataflowCanvas({ }; const isValidConnection = (connection: Connection | Edge): boolean => { - if (readOnly || !connection.source || !connection.target || connection.source === connection.target) return false; - const source = graph.nodes.find((node) => node.id === connection.source); - const target = graph.nodes.find((node) => node.id === connection.target); - if (!source || !target || source.type === "output" || target.type.startsWith("source.")) return false; - const sourcePort = connection.sourceHandle ?? "output"; - const targetPort = connection.targetHandle ?? "input"; - const sourceDefinition = definitions.get(source.type); - const targetDefinition = definitions.get(target.type); - if ( - !sourceDefinition?.output_ports.some((port) => port.id === sourcePort) - || !targetDefinition?.input_ports.some((port) => port.id === targetPort) - ) { - return false; - } - const duplicate = graph.edges.some( - (edge) => - edge.source === connection.source - && edge.target === connection.target - && (edge.source_port ?? "output") === sourcePort - && (edge.target_port ?? "input") === targetPort - ); - if (duplicate) return false; - const port = targetDefinition.input_ports.find((item) => item.id === targetPort); - const targetPortConnections = graph.edges.filter( - (edge) => edge.target === connection.target && (edge.target_port ?? "input") === targetPort - ); - if (!port?.multiple && targetPortConnections.length) return false; - return !wouldCreateCycle(graph, connection.source, connection.target); + if (readOnly || !connection.source || !connection.target) return false; + return definitionConnectionError( + graph, + nodeLibrary, + { + source: connection.source, + target: connection.target, + sourcePort: connection.sourceHandle, + targetPort: connection.targetHandle + } + ) === null; }; const onConnect = (connection: Connection) => { @@ -234,23 +218,6 @@ export default function DataflowCanvas({ ); } -function wouldCreateCycle(graph: PipelineGraph, source: string, target: string): boolean { - const outgoing = new Map(); - graph.edges.forEach((edge) => { - outgoing.set(edge.source, [...(outgoing.get(edge.source) ?? []), edge.target]); - }); - const pending = [target]; - const visited = new Set(); - while (pending.length) { - const nodeId = pending.pop(); - if (!nodeId || visited.has(nodeId)) continue; - if (nodeId === source) return true; - visited.add(nodeId); - pending.push(...(outgoing.get(nodeId) ?? [])); - } - return false; -} - export function updateGraphNode( graph: PipelineGraph, updatedNode: PipelineGraphNode diff --git a/webui/src/features/dataflow/DataflowPage.tsx b/webui/src/features/dataflow/DataflowPage.tsx index 6d2b7b3..5d81ec1 100644 --- a/webui/src/features/dataflow/DataflowPage.tsx +++ b/webui/src/features/dataflow/DataflowPage.tsx @@ -106,8 +106,8 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings const canImportSources = canWrite && sourceCatalogueWritable && ( - hasScope(auth, "connectors:source:write") - || hasScope(auth, "connectors:source:admin") + hasScope(auth, "datasources:stage:write") + || hasScope(auth, "datasources:source:admin") ); const dirty = Boolean(draft) && draftFingerprint(draft) !== draftFingerprint(savedDraft); const selectedNode = useMemo( @@ -571,7 +571,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings Nodes {canImportSources ? ( } variant="ghost" onClick={() => setSnapshotOpen(true)} @@ -815,7 +815,7 @@ function SourceSnapshotDialog({ return ( { if (!busy) onClose(); diff --git a/webui/src/features/dataflow/NodeInspector.tsx b/webui/src/features/dataflow/NodeInspector.tsx index 78020a3..df33c8a 100644 --- a/webui/src/features/dataflow/NodeInspector.tsx +++ b/webui/src/features/dataflow/NodeInspector.tsx @@ -151,7 +151,7 @@ export default function NodeInspector({ disabled={readOnly || !sourceCatalogueAvailable} > {sources.map((source) => (