feat: add reusable workflow definition graphs

src/govoplan_workflow/backend/validation.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+from govoplan_core.core.definition_graphs import (
+    DefinitionDiagnostic,
+    DefinitionEdge,
+    DefinitionNode,
+    validate_definition_graph,
+)
+from govoplan_workflow.backend.node_library import (
+    WORKFLOW_GRAPH_LIBRARY,
+    WORKFLOW_NODE_TYPES_BY_ID,
+)
+from govoplan_workflow.backend.schemas import WorkflowGraph
+
+
+def validate_workflow_graph(graph: WorkflowGraph) -> tuple[DefinitionDiagnostic, ...]:
+    diagnostics = list(
+        validate_definition_graph(
+            WORKFLOW_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
+            ),
+        )
+    )
+    outgoing = {node.id: 0 for node in graph.nodes}
+    for edge in graph.edges:
+        if edge.source in outgoing:
+            outgoing[edge.source] += 1
+    for node in graph.nodes:
+        definition = WORKFLOW_NODE_TYPES_BY_ID.get(node.type)
+        if definition is None:
+            continue
+        for field in definition.config_fields:
+            if field.required and _empty(node.config.get(field.id)):
+                diagnostics.append(
+                    DefinitionDiagnostic(
+                        severity="error",
+                        code="node.config_required",
+                        message=f"{definition.label} requires {field.label.lower()}.",
+                        node_id=node.id,
+                        field=field.id,
+                    )
+                )
+        if definition.output_ports and outgoing[node.id] == 0:
+            diagnostics.append(
+                DefinitionDiagnostic(
+                    severity="error",
+                    code="node.outgoing_required",
+                    message=f"{definition.label} must lead to another workflow step.",
+                    node_id=node.id,
+                )
+            )
+    return _deduplicate(diagnostics)
+
+
+def _empty(value: object) -> bool:
+    return value is None or value == "" or value == [] or value == {}
+
+
+def _deduplicate(
+    diagnostics: list[DefinitionDiagnostic],
+) -> tuple[DefinitionDiagnostic, ...]:
+    seen: set[tuple[str, str | None, str | None]] = set()
+    result: list[DefinitionDiagnostic] = []
+    for item in diagnostics:
+        key = (item.code, item.node_id, item.field)
+        if key in seen:
+            continue
+        seen.add(key)
+        result.append(item)
+    return tuple(result)
+
+
+__all__ = ["validate_workflow_graph"]