diff --git a/README.md b/README.md index b570bd1..c185277 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,34 @@ flow For property schemas and richer editors, provide `node_types`/`edge_types` with `PropertySpec` and handle changes via `ReactFlow.on(...)`. `.on` callbacks receive the event payload as the first argument and can optionally accept the `ReactFlow` instance as a second argument. +## Connection validation + +Connection checks are opt-in. For frontend-only validation, set `connection_validation` with any of `direction`, `types`, `capacity`, `duplicates`, or `cycles` set to `True`. `types` compares declared handle types case-insensitively and permits unknown types; `capacity` uses `maxConnections` on input handle dictionaries, with no limit when omitted. Existing handle connectability flags still apply. + +```python +from panel_reactflow import NodeType, ReactFlow + +flow = ReactFlow( + node_types={ + "source": NodeType(type="source", outputs=[{"id": "value", "type": "str"}]), + "sink": NodeType(type="sink", inputs=[{"id": "value", "type": "str", "maxConnections": 1}]), + }, + connection_validation={"direction": True, "types": True, "capacity": True, "cycles": True}, +) +``` + +For application rules, register a Python validator. On each drag start, ReactFlow requests results for every candidate port and shows the returned reasons while dragging. The callback receives an edge-shaped payload with `source`, `target`, `sourceHandle`, and `targetHandle`; return `None` to allow or a string to reject. Hooks and frontend checks can be used together. Validate again in your `edge_added` handler before accepting the connection, since the graph may have changed after drag start. + +```python +def validate_connection(edge, flow): + if edge["source"] == edge["target"]: + return "A node cannot connect to itself." + +flow.add_connection_validator(validate_connection) +``` + +Run the [connection validation demo](examples/connection_validation.py) with `PYTHONPATH=src pixi run panel serve examples/connection_validation.py --show` from the repository root. + ## Development ```bash diff --git a/docs/how-to/validate-connections.md b/docs/how-to/validate-connections.md new file mode 100644 index 0000000..bf46faa --- /dev/null +++ b/docs/how-to/validate-connections.md @@ -0,0 +1,82 @@ +# Validate Connections + +Use `connection_validation` to reject invalid connections as users drag between handles. The policies run in the browser; a Python validator can add application-specific rules. Both are opt-in and can be used together. + +## Configure browser policies + +Set only the checks your graph needs: + +| Policy | Rejects | +|--------|---------| +| `direction` | Connections that do not run from a declared output to a declared input. Users can still start a drag at either end. | +| `types` | Connections between handles with different declared `type` values (case-insensitive). If either type is missing, the connection is allowed. | +| `capacity` | Connections to an input handle that has reached its positive integer `maxConnections` limit. Inputs without a limit are unrestricted. | +| `duplicates` | A second edge with the same source, target, source handle, and target handle. Different handle pairs can still connect the same nodes. | +| `cycles` | Connections that would create a cycle, including a self-connection. | + +Policies default to off. Existing `NodeType` connectable flags still control whether a handle can start or receive a drag; see [Control Handle Connectivity](control-handle-connectivity.md). + +The following app uses handle types and capacity limits alongside a Python rule. Save it as `validation_app.py` and run `panel serve validation_app.py --show`: + +```python +import panel as pn + +from panel_reactflow import NodeSpec, NodeType, ReactFlow + +pn.extension("jsoneditor") + +flow = ReactFlow( + nodes=[ + NodeSpec(id="source", type="source", label="Source", position={"x": 0, "y": 0}), + NodeSpec(id="transform", type="transform", label="Transform", position={"x": 260, "y": 0}), + NodeSpec(id="publish", type="publish", label="Publish", position={"x": 520, "y": 0}), + ], + node_types={ + "source": NodeType(type="source", outputs=[{"id": "text", "type": "Text"}]), + "transform": NodeType( + type="transform", + inputs=[{"id": "text", "type": "text", "maxConnections": 1}], + outputs=[{"id": "cleaned", "type": "Text"}], + ), + "publish": NodeType(type="publish", inputs=[{"id": "text", "type": "Text", "maxConnections": 1}]), + }, + connection_validation={ + "direction": True, + "types": True, + "capacity": True, + "duplicates": True, + "cycles": True, + }, + height=350, + width=800, +) + + +def validate_connection(edge, flow): + if edge["source"] == "source" and edge["target"] == "publish": + return "Publish requires the Transform output." + return None + + +flow.add_connection_validator(validate_connection) + + +def on_edge_added(event, flow): + edge = event["edge"] + if validate_connection(edge, flow): + flow.remove_edge(edge["id"]) + + +flow.on("edge_added", on_edge_added) +flow.servable() +``` + +Drag from **Source.text** to **Transform.text**, then from **Transform.cleaned** to **Publish.text**. A direct connection from Source to Publish is rejected by Python. Try connecting Source.text to Transform.cleaned (two outputs) to see the direction check, or connect an output to an already occupied input to see the capacity check. + +## Add application rules + +Register callbacks with `flow.add_connection_validator(callback)` and unregister them with `flow.remove_connection_validator(callback)`. Each callback accepts either `edge` or `(edge, flow)` and returns `None` to allow the connection or a reason string to reject it. The payload has `source`, `target`, `sourceHandle`, and `targetHandle` keys; handles without an explicit ID use `None`. + +When a user starts dragging from an output or input, Python checks every candidate handle on the opposite side. Validators run in registration order until one rejects a candidate. Browser policies and Python reasons are shown on handles while dragging; the browser prevents a rejected connection. Python results must arrive before the connection is dropped: pending requests and requests taking longer than three seconds block the connection. Exceptions also reject the candidate and are logged on the server. Keep validators fast. + +Validation is for interactive drags, not a constraint on `flow.edges` or `flow.add_edge()`. Python validators run at drag start, not when the edge is added. If the rule depends on graph state that may change during the drag, check it again in an `edge_added` handler before keeping or persisting the edge. The example above removes an edge if its application rule no longer holds. diff --git a/docs/index.md b/docs/index.md index e28471f..3417c5f 100644 --- a/docs/index.md +++ b/docs/index.md @@ -62,6 +62,7 @@ flow.servable() - [Define Nodes & Edges](how-to/define-nodes-edges.md) - [Declare Node & Edge Types](how-to/declare-types.md) - [Control Handle Connectivity](how-to/control-handle-connectivity.md) — restrict connections +- [Validate Connections](how-to/validate-connections.md) — check types, capacity, cycles, and application rules - [Define Editors](how-to/define-editors.md) — node *and* edge editors - [Embed Views in Nodes](how-to/embed-views-in-nodes.md) - [Style Nodes & Edges](how-to/style-nodes-edges.md) diff --git a/examples/connection_validation.py b/examples/connection_validation.py new file mode 100644 index 0000000..6de56b1 --- /dev/null +++ b/examples/connection_validation.py @@ -0,0 +1,89 @@ +"""Try frontend rules and Python connection hooks in one graph. + +Run from the repository root with: + + PYTHONPATH=src pixi run panel serve examples/connection_validation.py --show +""" + +import panel as pn +import panel_material_ui as pmui + +from panel_reactflow import NodeSpec, NodeType, ReactFlow + +pn.extension() + + +class ConnectionValidationDemo(pn.viewable.Viewer): + def __init__(self, **params): + super().__init__(**params) + self._status = pn.pane.Markdown("No connections yet.") + self._flow = ReactFlow( + nodes=[ + NodeSpec(id="source", type="source", label="Source", position={"x": 0, "y": 60}).to_dict(), + NodeSpec(id="number", type="number", label="Number", position={"x": 0, "y": 280}).to_dict(), + NodeSpec(id="transform", type="transform", label="Transform", position={"x": 290, "y": 60}).to_dict(), + NodeSpec(id="publish", type="publish", label="Publish", position={"x": 610, "y": 60}).to_dict(), + NodeSpec(id="monitor", type="monitor", label="Monitor", position={"x": 610, "y": 280}).to_dict(), + ], + node_types={ + "source": NodeType(type="source", outputs=[{"id": "text", "type": "Text"}]), + "number": NodeType(type="number", outputs=[{"id": "value", "type": "Number"}]), + "transform": NodeType( + type="transform", + inputs=[{"id": "text", "type": "Text", "maxConnections": 1}], + outputs=[{"id": "cleaned", "type": "Text"}], + ), + "publish": NodeType(type="publish", inputs=[{"id": "text", "type": "Text", "maxConnections": 1}]), + "monitor": NodeType(type="monitor", inputs=[{"id": "text", "type": "Text"}]), + }, + connection_validation={ + "direction": True, + "types": True, + "capacity": True, + "duplicates": True, + "cycles": True, + }, + height=510, + sizing_mode="stretch_width", + ) + self._flow.add_connection_validator(self._validate_connection) + self._flow.on("edge_added", self._on_edge_added) + self._page = pmui.Page( + title="Connection validation", + main=[ + pmui.Container( + pmui.Column( + pn.pane.Markdown( + "Drag **Source.text** to **Transform.text**, then **Transform.cleaned** to **Publish.text**. " + "**Number.value** to **Transform.text** fails the frontend type check; " + "**Source.text** to **Publish.text** is rejected by Python. " + "Try connecting both text outputs to **Monitor.text**, or creating a cycle." + ), + self._flow, + self._status, + sizing_mode="stretch_width", + ), + width_option="lg", + ), + ], + ) + + def _validate_connection(self, edge, flow): + if edge["source"] == "source" and edge["target"] == "publish": + return "Publish requires text from Transform, not Source." + + def _on_edge_added(self, event, flow): + edge = event["edge"] + reason = self._validate_connection(edge, flow) + if reason: + flow.remove_edge(edge["id"]) + self._status.object = f"Rejected: {reason}" + else: + self._status.object = f"Connected: {edge['source']}.{edge['sourceHandle']} to {edge['target']}.{edge['targetHandle']}" + + def __panel__(self): + return self._page + + +demo = ConnectionValidationDemo() +demo.servable() diff --git a/src/panel_reactflow/base.py b/src/panel_reactflow/base.py index b4fa925..9958121 100644 --- a/src/panel_reactflow/base.py +++ b/src/panel_reactflow/base.py @@ -269,8 +269,9 @@ class NodeType: inputs : list of str or dict, optional List of input port definitions. Each entry can be a plain string (the handle ID) or a dict with ``"id"`` and optional ``"label"`` - and ``"type"`` keys. When a label and/or type is provided it renders - as a tooltip on hover, e.g. ``"Raw data input (DataFrame)"``. + and ``"type"`` keys. An optional ``"maxConnections"`` integer limits + connections when ``connection_validation["capacity"]`` is enabled. + Labels and types render as tooltips on hover. outputs : list of str or dict, optional List of output port definitions. Each entry can be a plain string (the handle ID) or a dict with ``"id"`` and optional ``"label"`` @@ -361,8 +362,8 @@ class NodeType: type: str label: str | None = None schema: Any = None - inputs: list[str | dict[str, str]] | None = None - outputs: list[str | dict[str, str]] | None = None + inputs: list[str | dict[str, Any]] | None = None + outputs: list[str | dict[str, Any]] | None = None input_connectable: bool = True input_connectable_start: bool = True input_connectable_end: bool = True @@ -1525,6 +1526,8 @@ class ReactFlow(ReactComponent): edges = param.List(default=[], doc="Canonical list of edge dictionaries or Edge instances.") node_types = param.Dict(default={}, doc="Node type descriptors keyed by type name.") edge_types = param.Dict(default={}, doc="Edge type descriptors keyed by type name.") + connection_validation = param.Dict(default={}, doc="Opt-in frontend connection policies (direction, cycles, duplicates, types, capacity).") + has_connection_validators = param.Boolean(default=False, doc="Whether Python connection validators are registered.") node_editors = param.Dict(default={}, doc="Node editor factories keyed by type name.", precedence=-1) edge_editors = param.Dict(default={}, doc="Edge editor factories keyed by type name.", precedence=-1) @@ -1645,6 +1648,7 @@ def __init__(self, **params: Any): params["edges"] = [ReactFlow._coerce_edge(edge) for edge in params["edges"]] super().__init__(**params) self._event_handlers: dict[str, list[Callable]] = {"*": []} + self._connection_validators: list[Callable] = [] self.param.watch(self._sync_instance_flow_refs, ["nodes", "edges"]) self.param.watch(self._normalize_nodes, ["nodes"]) self.param.watch(self._normalize_edges, ["edges"]) @@ -2448,6 +2452,8 @@ def _handle_msg(self, msg: dict[str, Any]) -> None: def _process_msg(self, msg: dict[str, Any]) -> None: match msg.get("type"): + case "connection_validation_requested": + self._validate_connection_request(msg) case "sync": nodes = msg.get("nodes") edges = msg.get("edges") @@ -2542,6 +2548,60 @@ def _process_msg(self, msg: dict[str, Any]) -> None: case _: return + def add_connection_validator(self, callback: Callable) -> None: + """Register a validator returning None to allow or a reason to reject. + + Callbacks accept either a connection payload or (payload, flow). + They run in registration order for each candidate during a drag. + """ + if not callable(callback): + raise TypeError("Connection validator must be callable.") + self._connection_validators.append(callback) + self.has_connection_validators = True + + def remove_connection_validator(self, callback: Callable) -> None: + """Unregister a previously added connection validator.""" + self._connection_validators.remove(callback) + self.has_connection_validators = bool(self._connection_validators) + + def _validate_connection_request(self, msg: dict[str, Any]) -> None: + node_id = msg["node_id"] + handle_id = msg["handle_id"] + handle_type = msg["handle_type"] + opposite_type = "target" if handle_type == "source" else "source" + port_key = "inputs" if opposite_type == "target" else "outputs" + results = [] + for node in self.nodes: + candidate_id = self._node_id(node) + node_type = self.node_types.get(self._node_type(node), {}) + handles = node_type.get(port_key) + if handles is None: + handles = [None] + for handle in handles: + candidate_handle = handle.get("id") if isinstance(handle, dict) else handle + if handle_type == "source": + payload = {"source": node_id, "target": candidate_id, "sourceHandle": handle_id, "targetHandle": candidate_handle} + else: + payload = {"source": candidate_id, "target": node_id, "sourceHandle": candidate_handle, "targetHandle": handle_id} + reason = None + for callback in self._connection_validators: + try: + if len(inspect.signature(callback).parameters) == 2: + reason = callback(payload, self) + else: + reason = callback(payload) + if reason is not None and not isinstance(reason, str): + reason = "Connection validator must return None or a reason string." + elif reason == "": + reason = "Connection rejected." + except Exception as exc: + _LOGGER.exception("Connection validator failed") + reason = str(exc) or "Connection validator failed." + if reason is not None: + break + results.append({"node_id": candidate_id, "handle_id": candidate_handle, "handle_type": opposite_type, "reason": reason}) + self._send_msg({"type": "connection_validation_result", "request_id": msg["request_id"], "results": results}) + def _handle_client_error(self, msg: dict[str, Any]) -> None: """Log a client-side error reported by the frontend and re-emit it. diff --git a/src/panel_reactflow/dist/css/reactflow.css b/src/panel_reactflow/dist/css/reactflow.css index 704730b..31d9347 100644 --- a/src/panel_reactflow/dist/css/reactflow.css +++ b/src/panel_reactflow/dist/css/reactflow.css @@ -107,6 +107,35 @@ opacity: 1; } +.react-flow__handle.rf-handle-valid { + background: #16803c; + border-color: #16803c; +} + +.react-flow__handle.rf-handle-pending { + background: #797d87; + border-color: #797d87; +} + +.react-flow__handle.rf-handle-invalid { + background: #bb3b35; + border-color: #bb3b35; +} + +.rf-validation-status { + position: absolute; + bottom: 12px; + left: 50%; + transform: translateX(-50%); + padding: 6px 10px; + border-radius: 4px; + background: var(--panel-background-color, #fff); + color: var(--panel-on-background-color, #222); + box-shadow: 0 2px 8px rgba(0, 0, 0, 0.2); + pointer-events: none; + z-index: 5; +} + .rf-context-menu { background: var(--xy-node-background-color, var(--panel-background-color)); border: 1px solid var(--panel-border-color); diff --git a/src/panel_reactflow/models/reactflow.jsx b/src/panel_reactflow/models/reactflow.jsx index 249d841..b7abf7a 100644 --- a/src/panel_reactflow/models/reactflow.jsx +++ b/src/panel_reactflow/models/reactflow.jsx @@ -22,7 +22,65 @@ const RETRY_DELAY_MS = 100; // How long a remounted flow must survive before its retry budget is refilled. const HEALTHY_RESET_MS = 5000; const HOVER_CLOSE_GRACE_MS = 150; +const CONNECTION_VALIDATION_TIMEOUT_MS = 3000; const popupHoverState = { inside: false, closeTimers: new Set() }; +const ConnectionValidationContext = React.createContext(null); + +function connectionKey(nodeId, handleId, handleType) { + return JSON.stringify([nodeId, handleId ?? null, handleType]); +} + +function getHandle(spec, direction, id) { + const handles = spec?.[direction === "source" ? "outputs" : "inputs"]; + if (handles == null) return id == null ? {} : null; + return handles.find((handle) => (typeof handle === "string" ? handle : handle.id) === id) ?? null; +} + +function localConnectionReason(connection, nodes, edges, nodeTypeSpecs, policy) { + if (!Object.values(policy || {}).some(Boolean)) return null; + const source = nodes.find((node) => node.id === connection.source); + const target = nodes.find((node) => node.id === connection.target); + if (!source || !target) return "Node not found"; + const sourceHandle = getHandle(nodeTypeSpecs[source.type], "source", connection.sourceHandle); + const targetHandle = getHandle(nodeTypeSpecs[target.type], "target", connection.targetHandle); + if (policy.direction && (!sourceHandle || !targetHandle)) return "Connect an output to an input"; + if (policy.cycles) { + if (source.id === target.id) return "Connection would create a cycle"; + const successors = new Map(); + edges.forEach((edge) => { + if (!successors.has(edge.source)) successors.set(edge.source, []); + successors.get(edge.source).push(edge.target); + }); + const visited = new Set(); + const queue = [target.id]; + while (queue.length) { + const current = queue.pop(); + if (current === source.id) return "Connection would create a cycle"; + if (visited.has(current)) continue; + visited.add(current); + queue.push(...(successors.get(current) || [])); + } + } + if (policy.duplicates && edges.some((edge) => + edge.source === source.id && edge.target === target.id && + (edge.sourceHandle ?? null) === (connection.sourceHandle ?? null) && + (edge.targetHandle ?? null) === (connection.targetHandle ?? null) + )) return "Connection already exists"; + if (policy.capacity && targetHandle && typeof targetHandle === "object") { + const limit = targetHandle.maxConnections; + if (Number.isInteger(limit) && limit > 0 && + edges.filter((edge) => edge.target === target.id && (edge.targetHandle ?? null) === (connection.targetHandle ?? null)).length >= limit + ) return "Input already has a connection"; + } + if (policy.types && sourceHandle && targetHandle && typeof sourceHandle === "object" && typeof targetHandle === "object") { + const sourceType = sourceHandle.type; + const targetType = targetHandle.type; + if (sourceType && targetType && String(sourceType).toLowerCase() !== String(targetType).toLowerCase()) { + return `Type mismatch: ${sourceType} → ${targetType}`; + } + } + return null; +} function schedulePopupClose(callback) { const timer = setTimeout(() => { @@ -72,6 +130,9 @@ function renderHandles(direction, handles, opts = {}) { const tooltipPos = direction === "input" ? "left" : "right"; const onHandleClick = opts.onHandleClick; const onHandleHover = opts.onHandleHover; + const validation = opts.validation; + const cannotReceive = validation?.drag && validation.drag.handleType !== handleType && + (opts.connectable === false || opts.connectableEnd === false); const makeHoverHandler = (id, eventType) => onHandleHover @@ -106,6 +167,7 @@ function renderHandles(direction, handles, opts = {}) { } // null/undefined → default handle if (!handles?.length) { + const reason = cannotReceive ? "Handle cannot accept connections" : validation?.reasonForHandle(opts.nodeId, null, handleType); return ( ); @@ -123,6 +188,7 @@ function renderHandles(direction, handles, opts = {}) { const label = typeof handle === "object" ? handle.label : undefined; const dtype = typeof handle === "object" ? handle.type : undefined; const tooltip = label && dtype ? `${label} (${dtype})` : label || dtype; + const reason = cannotReceive ? "Handle cannot accept connections" : validation?.reasonForHandle(opts.nodeId, id, handleType); return ( )} {renderHandles("input", spec.inputs, { - connectable: spec.inputConnectable, + nodeId: id, + validation, + connectable: isConnectable === false ? false : spec.inputConnectable, connectableStart: spec.inputConnectableStart, connectableEnd: spec.inputConnectableEnd, onHandleClick, @@ -346,7 +417,9 @@ function makeNodeComponent(typeName, typeSpec, editorMode, model, valuePopupTrig )} {renderHandles("output", spec.outputs, { - connectable: spec.outputConnectable, + nodeId: id, + validation, + connectable: isConnectable === false ? false : spec.outputConnectable, connectableStart: spec.outputConnectableStart, connectableEnd: spec.outputConnectableEnd, onHandleClick, @@ -660,11 +733,14 @@ function FlowInner({ onPaneClick, defaultEdgeOptions, nodeTypes, + nodeTypeSpecs, edgeTypes, nodeEditors, colorMode, editable, enableConnect, + connectionValidation, + hasConnectionValidators, enableDelete, enableMultiselect, maxZoom, @@ -685,12 +761,89 @@ function FlowInner({ const lastHydrated = useRef({ nodeRevision: null, nodesSig: null, edgesSig: null }); const lastViewportSig = useRef(null); const { setViewport: setRfViewport } = useReactFlow(); + const [drag, setDrag] = useState(null); + const [validationResults, setValidationResults] = useState(null); + const dragRef = useRef(null); + const resultsRef = useRef(null); + const validationTimerRef = useRef(null); + const requestIdRef = useRef(0); + + const candidateReason = useCallback((connection) => { + const local = localConnectionReason(connection, nodesRef.current, edgesRef.current, nodeTypeSpecs, connectionValidation); + if (local) return local; + if (!hasConnectionValidators) return null; + const current = dragRef.current; + if (!current) return "Connection validation unavailable"; + const result = resultsRef.current; + if (!result || result.requestId !== current.requestId) return "Checking connections..."; + if (result.error) return result.error; + const end = current.handleType === "source" + ? connectionKey(connection.target, connection.targetHandle, "target") + : connectionKey(connection.source, connection.sourceHandle, "source"); + if (!result.reasons.has(end)) return "Connection not validated"; + return result.reasons.get(end) === "" ? "Connection rejected" : result.reasons.get(end); + }, [connectionValidation, hasConnectionValidators, nodeTypeSpecs]); + + const reasonForHandle = useCallback((nodeId, handleId, handleType) => { + if (!drag || drag.handleType === handleType) return null; + const connection = drag.handleType === "source" + ? { source: drag.nodeId, sourceHandle: drag.handleId, target: nodeId, targetHandle: handleId } + : { source: nodeId, sourceHandle: handleId, target: drag.nodeId, targetHandle: drag.handleId }; + return candidateReason(connection); + }, [candidateReason, drag, validationResults]); + + const onConnectStart = useCallback((_event, { nodeId, handleId, handleType }) => { + if (!hasConnectionValidators && !Object.values(connectionValidation || {}).some(Boolean)) return; + const next = { nodeId, handleId: handleId ?? null, handleType, requestId: ++requestIdRef.current }; + dragRef.current = next; + resultsRef.current = null; + setDrag(next); + setValidationResults(null); + if (validationTimerRef.current) clearTimeout(validationTimerRef.current); + if (hasConnectionValidators) { + model.send_msg({ + type: "connection_validation_requested", request_id: next.requestId, + node_id: nodeId, handle_id: handleId ?? null, handle_type: handleType, + }); + validationTimerRef.current = setTimeout(() => { + if (dragRef.current?.requestId !== next.requestId) return; + dragRef.current = { ...dragRef.current, expired: true }; + const failure = { requestId: next.requestId, reasons: new Map(), error: "Connection validation timed out" }; + resultsRef.current = failure; + setValidationResults(failure); + }, CONNECTION_VALIDATION_TIMEOUT_MS); + } + }, [connectionValidation, hasConnectionValidators, model]); + + const onConnectEnd = useCallback(() => { + if (validationTimerRef.current) clearTimeout(validationTimerRef.current); + validationTimerRef.current = null; + dragRef.current = null; + resultsRef.current = null; + setDrag(null); + setValidationResults(null); + }, []); useEffect(() => { const handler = (msg) => { if (!msg || typeof msg !== "object") { return; } + if (msg.type === "connection_validation_result") { + if (dragRef.current?.requestId !== msg.request_id || dragRef.current.expired) return; + if (validationTimerRef.current) clearTimeout(validationTimerRef.current); + validationTimerRef.current = null; + const result = { + requestId: msg.request_id, + reasons: new Map((msg.results || []).map((entry) => [ + connectionKey(entry.node_id, entry.handle_id, entry.handle_type), entry.reason ?? null, + ])), + error: msg.error || null, + }; + resultsRef.current = result; + setValidationResults(result); + return; + } if (msg.type === "patch_node_data") { setNodes((current) => current.map((node) => { @@ -750,6 +903,10 @@ function FlowInner({ }; }, [model, setEdges, setNodes]); + useEffect(() => () => { + if (validationTimerRef.current) clearTimeout(validationTimerRef.current); + }, []); + useEffect(() => { nodesRef.current = nodes; }, [nodes]); @@ -845,16 +1002,20 @@ function FlowInner({ const onConnect = useCallback( (connection) => { - if (!enableConnect) { + if (!enableConnect || candidateReason(connection)) { return; } - const edgeId = connection.id || `${connection.source}->${connection.target}`; + const baseId = connection.id || `${connection.source}->${connection.target}`; + let edgeId = baseId; + let suffix = 1; + while (edgesRef.current.some((edge) => edge.id === edgeId)) edgeId = `${baseId}:${suffix++}`; const newEdge = { ...connection, id: edgeId }; const updated = addEdge(newEdge, edgesRef.current); + if (updated === edgesRef.current) return; setEdges(updated); sendPatch({ type: "edge_added", edge: newEdge }); }, - [enableConnect, sendPatch, setEdges], + [candidateReason, enableConnect, sendPatch, setEdges], ); const handleNodesChange = useCallback( @@ -1027,6 +1188,8 @@ function FlowInner({ onNodesDelete: wrap("onNodesDelete", onNodesDelete), onEdgesDelete: wrap("onEdgesDelete", onEdgesDelete), onConnect: wrap("onConnect", onConnect), + onConnectStart: wrap("onConnectStart", onConnectStart), + onConnectEnd: wrap("onConnectEnd", onConnectEnd), onMoveEnd: wrap("onMoveEnd", onMoveEnd), onNodeDoubleClick: wrap("onNodeDoubleClick", onNodeDoubleClick), onNodeContextMenu: wrap("onNodeContextMenu", onNodeContextMenu), @@ -1038,6 +1201,8 @@ function FlowInner({ }, [ handleNodesChange, onConnect, + onConnectStart, + onConnectEnd, onEdgeClick, onEdgeHover, onEdgeHoverEnd, @@ -1052,7 +1217,9 @@ function FlowInner({ reportError, ]); + const validationEnabled = hasConnectionValidators || Object.values(connectionValidation || {}).some(Boolean); return ( + !candidateReason(connection) : undefined} nodesDraggable={editable} nodesConnectable={editable && enableConnect} elementsSelectable={editable} @@ -1074,6 +1242,12 @@ function FlowInner({ {showMinimap ? : null} + {drag && hasConnectionValidators && (!validationResults || validationResults.error) ? ( +
+ {validationResults?.error || "Checking connections..."} +
+ ) : null} + ); } @@ -1095,6 +1269,8 @@ export function render({ model, view }) { const [hoverDelay] = model.useState("popup_hover_delay"); const [errorRecovery] = model.useState("error_recovery"); const [enableConnect] = model.useState("enable_connect"); + const [connectionValidation] = model.useState("connection_validation"); + const [hasConnectionValidators] = model.useState("has_connection_validators"); const [enableDelete] = model.useState("enable_delete"); const [enableMultiselect] = model.useState("enable_multiselect"); const [maxZoom] = model.useState("max_zoom"); @@ -1397,10 +1573,13 @@ export function render({ model, view }) { defaultEdgeOptions={defaultEdgeOptions} colorMode={colorMode} nodeTypes={hydratedNodeTypes} + nodeTypeSpecs={allNodeTypes} edgeTypes={hydratedEdgeTypes} nodeEditors={nodeEditors} editable={editable} enableConnect={enableConnect} + connectionValidation={connectionValidation} + hasConnectionValidators={hasConnectionValidators} enableDelete={enableDelete} enableMultiselect={enableMultiselect} maxZoom={maxZoom} diff --git a/tests/test_api.py b/tests/test_api.py index 2ddfe6c..b9a54cc 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -463,6 +463,185 @@ def _two_node_flow(**params) -> ReactFlow: ) +def test_connection_validator_registration_and_policy_sync() -> None: + """Registration controls the synced hook flag; policies remain independently configurable.""" + policy = {"direction": True, "cycles": True, "duplicates": False, "types": True, "capacity": True} + flow = ReactFlow(connection_validation=policy) + first = lambda payload: None + second = lambda payload, flow: None + + assert flow.has_connection_validators is False + assert flow._process_param_change({"connection_validation": flow.connection_validation})["connection_validation"] == policy + assert flow._process_param_change({"has_connection_validators": False})["has_connection_validators"] is False + flow.add_connection_validator(first) + flow.add_connection_validator(second) + assert flow.has_connection_validators is True + flow.remove_connection_validator(first) + assert flow.has_connection_validators is True + flow.remove_connection_validator(second) + assert flow.has_connection_validators is False + with pytest.raises(TypeError, match="callable"): + flow.add_connection_validator(None) + + +def test_connection_validator_flag_syncs_to_model(document, comm) -> None: + """The browser receives both the registration flag and opt-in policy updates.""" + flow = ReactFlow() + model = flow.get_root(document, comm=comm) + assert model.data.has_connection_validators is False + assert model.data.connection_validation == {} + + def validate(payload): + return None + + flow.add_connection_validator(validate) + flow.connection_validation = {"cycles": True} + assert model.data.has_connection_validators is True + assert model.data.connection_validation == {"cycles": True} + flow.remove_connection_validator(validate) + assert model.data.has_connection_validators is False + + +def test_connection_validation_source_drag_includes_all_target_handles() -> None: + """A source drag checks target ports on every node, including itself and defaults.""" + flow = ReactFlow( + nodes=[ + Node(id="start", type="typed"), + {"id": "plain", "type": "panel"}, + {"id": "empty", "type": "empty"}, + ], + node_types={ + "typed": NodeType(type="typed", inputs=["in", {"id": "other", "maxConnections": 2}]), + "panel": {"inputs": None, "outputs": None}, + "empty": NodeType(type="empty", inputs=[]), + }, + ) + msgs = _capture_msgs(flow) + flow._handle_msg({"type": "connection_validation_requested", "request_id": 12, "node_id": "start", "handle_id": "out", "handle_type": "source"}) + + assert msgs == [ + { + "type": "connection_validation_result", + "request_id": 12, + "results": [ + {"node_id": "start", "handle_id": "in", "handle_type": "target", "reason": None}, + {"node_id": "start", "handle_id": "other", "handle_type": "target", "reason": None}, + {"node_id": "plain", "handle_id": None, "handle_type": "target", "reason": None}, + ], + } + ] + + +def test_connection_validation_reverse_drag_passes_oriented_payload() -> None: + """A target drag builds canonical source-to-target payloads for candidate outputs.""" + flow = ReactFlow( + nodes=[{"id": "start", "type": "typed"}, {"id": "plain"}, {"id": "empty", "type": "empty"}], + node_types={"typed": NodeType(type="typed", outputs=["out", {"id": "other"}]), "empty": NodeType(type="empty", outputs=[])}, + ) + seen = [] + + def validate(payload, graph): + assert graph is flow + seen.append(payload.copy()) + return None + + flow.add_connection_validator(validate) + msgs = _capture_msgs(flow) + flow._handle_msg({"type": "connection_validation_requested", "request_id": 13, "node_id": "start", "handle_id": "in", "handle_type": "target"}) + + assert seen == [ + {"source": "start", "target": "start", "sourceHandle": "out", "targetHandle": "in"}, + {"source": "start", "target": "start", "sourceHandle": "other", "targetHandle": "in"}, + {"source": "plain", "target": "start", "sourceHandle": None, "targetHandle": "in"}, + ] + assert msgs == [ + { + "type": "connection_validation_result", + "request_id": 13, + "results": [ + {"node_id": "start", "handle_id": "out", "handle_type": "source", "reason": None}, + {"node_id": "start", "handle_id": "other", "handle_type": "source", "reason": None}, + {"node_id": "plain", "handle_id": None, "handle_type": "source", "reason": None}, + ], + } + ] + + +def test_connection_validation_rejection_and_callback_order() -> None: + """The first rejecting hook supplies the reason; later hooks skip only that candidate.""" + flow = _two_node_flow() + seen = [] + + def reject_self(payload): + seen.append(("first", payload["target"])) + return "No loops" if payload["source"] == payload["target"] else None + + def reject_other(payload, graph): + assert graph is flow + seen.append(("second", payload["target"])) + return "No n2" if payload["target"] == "n2" else None + + flow.add_connection_validator(reject_self) + flow.add_connection_validator(reject_other) + msgs = _capture_msgs(flow) + flow._handle_msg({"type": "connection_validation_requested", "request_id": 14, "node_id": "n1", "handle_id": None, "handle_type": "source"}) + + assert seen == [("first", "n1"), ("first", "n2"), ("second", "n2")] + assert [result["reason"] for result in msgs[0]["results"]] == ["No loops", "No n2"] + + +def test_empty_validator_reason_still_rejects() -> None: + flow = _two_node_flow() + flow.add_connection_validator(lambda payload: "") + msgs = _capture_msgs(flow) + + flow._handle_msg({"type": "connection_validation_requested", "request_id": 18, "node_id": "n1", "handle_id": None, "handle_type": "source"}) + + assert all(result["reason"] == "Connection rejected." for result in msgs[0]["results"]) + + +def test_connection_validators_only_run_on_drag_requests() -> None: + """Normal edge additions do not call connection validators or emit validation replies.""" + flow = _two_node_flow() + seen = [] + + def validate(payload): + seen.append(payload) + return "blocked" + + flow.add_connection_validator(validate) + msgs = _capture_msgs(flow) + flow._handle_msg({"type": "edge_added", "edge": {"id": "e1", "source": "n1", "target": "n2"}}) + + assert len(flow.edges) == 1 + assert seen == [] + assert msgs == [] + + +@pytest.mark.parametrize("error", [ValueError("broken"), ValueError("")]) +def test_connection_validation_hook_errors_fail_closed(error, caplog) -> None: + """An exception rejects a candidate and still produces a response for every handle.""" + flow = _two_node_flow() + + def raises(payload): + raise error + + flow.add_connection_validator(raises) + msgs = _capture_msgs(flow) + flow._handle_msg({"type": "connection_validation_requested", "request_id": 15, "node_id": "n1", "handle_id": None, "handle_type": "source"}) + + assert [result["reason"] for result in msgs[0]["results"]] == [str(error) or "Connection validator failed."] * 2 + assert "Connection validator failed" in caplog.text + + +def test_connection_validation_no_candidates_still_replies() -> None: + """Empty port lists produce an empty correlated response without invoking hooks.""" + flow = ReactFlow(nodes=[{"id": "n1", "type": "empty"}], node_types={"empty": NodeType(type="empty", inputs=[])}) + msgs = _capture_msgs(flow) + flow._handle_msg({"type": "connection_validation_requested", "request_id": 16, "node_id": "n1", "handle_id": None, "handle_type": "source"}) + assert msgs == [{"type": "connection_validation_result", "request_id": 16, "results": []}] + + def test_node_base_param_change_patches_props() -> None: node = Node(id="n1", position={"x": 0, "y": 0}, label="A") flow = ReactFlow(nodes=[node]) diff --git a/tests/ui/test_ui.py b/tests/ui/test_ui.py index 6beb28a..8df8f30 100644 --- a/tests/ui/test_ui.py +++ b/tests/ui/test_ui.py @@ -1,6 +1,10 @@ """UI tests for ReactFlow using Playwright.""" import re +import runpy +import threading +import time +from pathlib import Path import panel as pn import panel.models.jsoneditor # noqa @@ -167,6 +171,7 @@ def _viewport_updated(): def test_connecting_edge_updates_python(page): + """Repeated drags without duplicate validation keep browser and Python edges in sync.""" flow = _make_flow(include_edge=False) serve_component(page, flow) @@ -181,6 +186,238 @@ def _edge_added(): wait_until(_edge_added, timeout=8000) expect(page.locator(".react-flow__edge")).to_have_count(1) + source_handle.drag_to(target_handle) + page.wait_for_timeout(300) + expect(page.locator(".react-flow__edge")).to_have_count(1) + assert len(flow.edges) == 1 + + +def test_frontend_connection_validation_filters_targets(page): + flow = ReactFlow( + nodes=[ + NodeSpec(id="source", type="producer", label="Source", position={"x": 0, "y": 0}).to_dict(), + NodeSpec(id="wrong", type="wrong", label="Wrong", position={"x": 260, "y": 0}).to_dict(), + NodeSpec(id="right", type="right", label="Right", position={"x": 260, "y": 180}).to_dict(), + ], + node_types={ + "producer": NodeType(type="producer", outputs=[{"id": "out", "type": "Text"}]), + "wrong": NodeType(type="wrong", inputs=[{"id": "in", "type": "Number"}]), + "right": NodeType(type="right", inputs=[{"id": "in", "type": "text", "maxConnections": 1}]), + }, + connection_validation={"direction": True, "types": True, "capacity": True}, + width=750, + height=500, + ) + serve_component(page, flow) + source = _node_locator(page, "Source").locator(".react-flow__handle-right") + wrong = _node_locator(page, "Wrong").locator(".react-flow__handle-left") + right = _node_locator(page, "Right").locator(".react-flow__handle-left") + + source.drag_to(wrong) + expect(page.locator(".react-flow__edge")).to_have_count(0) + assert flow.edges == [] + + source.drag_to(right) + wait_until(lambda: len(flow.edges) == 1, timeout=8000) + source.drag_to(right) + expect(page.locator(".react-flow__edge")).to_have_count(1) + assert len(flow.edges) == 1 + + +def test_frontend_connection_validation_blocks_duplicates_and_cycles(page): + flow = ReactFlow( + nodes=[ + NodeSpec(id="a", type="step", label="A", position={"x": 0, "y": 0}).to_dict(), + NodeSpec(id="b", type="step", label="B", position={"x": 260, "y": 0}).to_dict(), + ], + node_types={"step": NodeType(type="step", inputs=["in"], outputs=["out"])}, + connection_validation={"direction": True, "cycles": True, "duplicates": True}, + width=650, + height=350, + ) + serve_component(page, flow) + output_a = _node_locator(page, "A").locator(".react-flow__handle-right") + input_a = _node_locator(page, "A").locator(".react-flow__handle-left") + output_b = _node_locator(page, "B").locator(".react-flow__handle-right") + input_b = _node_locator(page, "B").locator(".react-flow__handle-left") + + output_a.drag_to(input_b) + wait_until(lambda: len(flow.edges) == 1, timeout=8000) + output_a.drag_to(input_b) + output_b.drag_to(input_a) + expect(page.locator(".react-flow__edge")).to_have_count(1) + assert len(flow.edges) == 1 + + +def test_frontend_validation_supports_multiple_ports_between_nodes(page): + flow = ReactFlow( + nodes=[ + NodeSpec(id="source", type="producer", label="Source", position={"x": 0, "y": 0}).to_dict(), + NodeSpec(id="target", type="consumer", label="Target", position={"x": 260, "y": 0}).to_dict(), + ], + node_types={ + "producer": NodeType(type="producer", outputs=["a", "b"]), + "consumer": NodeType(type="consumer", inputs=["a", "b"]), + }, + connection_validation={"direction": True, "duplicates": True}, + width=650, + height=350, + ) + serve_component(page, flow) + source = _node_locator(page, "Source").locator(".react-flow__handle-right") + target = _node_locator(page, "Target").locator(".react-flow__handle-left") + + source.nth(0).drag_to(target.nth(0)) + wait_until(lambda: len(flow.edges) == 1, timeout=8000) + source.nth(1).drag_to(target.nth(1)) + wait_until(lambda: len(flow.edges) == 2, timeout=8000) + assert {edge["id"] for edge in flow.edges} == {"source->target", "source->target:1"} + + +def test_disabled_handle_end_is_not_highlighted_as_valid(page): + flow = ReactFlow( + nodes=[ + NodeSpec(id="source", type="producer", label="Source", position={"x": 0, "y": 0}).to_dict(), + NodeSpec(id="sink", type="consumer", label="Sink", position={"x": 260, "y": 0}).to_dict(), + ], + node_types={ + "producer": NodeType(type="producer", outputs=["out"], output_connectable_end=False), + "consumer": NodeType(type="consumer", inputs=["in"], input_connectable_end=False), + }, + connection_validation={"direction": True}, + width=650, + height=350, + ) + serve_component(page, flow) + source = _node_locator(page, "Source").locator(".react-flow__handle-right") + sink = _node_locator(page, "Sink").locator(".react-flow__handle-left") + + for origin, destination in ((source, sink), (sink, source)): + box = origin.bounding_box() + page.mouse.move(box["x"] + box["width"] / 2, box["y"] + box["height"] / 2) + page.mouse.down() + page.mouse.move(box["x"] + box["width"] / 2 + 25, box["y"] + box["height"] / 2) + expect(destination).to_have_class(re.compile("rf-handle-invalid")) + hint = destination.get_attribute("title") or destination.get_attribute("data-tooltip") + assert hint == "Handle cannot accept connections" + page.mouse.up() + + +def test_python_connection_validation_filters_targets(page): + flow = ReactFlow( + nodes=[ + NodeSpec(id="source", type="producer", label="Source", position={"x": 0, "y": 0}).to_dict(), + NodeSpec(id="wrong", type="consumer", label="Wrong", position={"x": 260, "y": 0}).to_dict(), + NodeSpec(id="right", type="consumer", label="Right", position={"x": 260, "y": 180}).to_dict(), + ], + node_types={ + "producer": NodeType(type="producer", outputs=["out"]), + "consumer": NodeType(type="consumer", inputs=["in"]), + }, + width=750, + height=500, + ) + flow.add_connection_validator(lambda payload, _flow: "Blocked by Python" if payload["target"] == "wrong" else None) + serve_component(page, flow) + source = _node_locator(page, "Source").locator(".react-flow__handle-right") + wrong = _node_locator(page, "Wrong").locator(".react-flow__handle-left") + right = _node_locator(page, "Right").locator(".react-flow__handle-left") + + box = source.bounding_box() + page.mouse.move(box["x"] + box["width"] / 2, box["y"] + box["height"] / 2) + page.mouse.down() + page.mouse.move(box["x"] + box["width"] / 2 + 25, box["y"] + box["height"] / 2) + expect(wrong).to_have_class(re.compile("rf-handle-invalid")) + expect(right).to_have_class(re.compile("rf-handle-valid")) + expect(wrong).to_have_attribute("data-tooltip", "Blocked by Python") + page.mouse.up() + + source.drag_to(wrong) + expect(page.locator(".react-flow__edge")).to_have_count(0) + box = right.bounding_box() + destination = source.bounding_box() + page.mouse.move(box["x"] + box["width"] / 2, box["y"] + box["height"] / 2) + page.mouse.down() + page.mouse.move(box["x"] + box["width"] / 2 - 25, box["y"] + box["height"] / 2) + expect(source).to_have_class(re.compile("rf-handle-valid")) + page.mouse.move(destination["x"] + destination["width"] / 2, destination["y"] + destination["height"] / 2) + page.mouse.up() + wait_until(lambda: len(flow.edges) == 1, timeout=8000) + assert flow.edges[0]["source"] == "source" and flow.edges[0]["target"] == "right" + + +def test_late_python_validation_reply_does_not_reopen_drag(page): + flow = ReactFlow( + nodes=[ + NodeSpec(id="source", label="Source", position={"x": 0, "y": 0}).to_dict(), + NodeSpec(id="target", label="Target", position={"x": 260, "y": 0}).to_dict(), + ], + width=650, + height=350, + ) + + completed = threading.Event() + + def validate(payload): + if payload["target"] == "target": + time.sleep(3.5) + completed.set() + + flow.add_connection_validator(validate) + serve_component(page, flow) + source = _node_locator(page, "Source").locator(".react-flow__handle-right") + target = _node_locator(page, "Target").locator(".react-flow__handle-left") + origin = source.bounding_box() + page.mouse.move(origin["x"] + origin["width"] / 2, origin["y"] + origin["height"] / 2) + page.mouse.down() + page.mouse.move(origin["x"] + origin["width"] / 2 + 25, origin["y"] + origin["height"] / 2) + expect(page.locator(".rf-validation-status")).to_have_text("Connection validation timed out") + wait_until(completed.is_set, timeout=8000) + page.wait_for_timeout(300) + expect(target).to_have_attribute("title", "Connection validation timed out", timeout=8000) + destination = target.bounding_box() + page.mouse.move(destination["x"] + destination["width"] / 2, destination["y"] + destination["height"] / 2) + page.mouse.up() + expect(page.locator(".react-flow__edge")).to_have_count(0) + assert flow.edges == [] + + +def test_connection_validation_demo(page): + demo_path = Path(__file__).resolve().parents[2] / "examples" / "connection_validation.py" + demo = runpy.run_path(str(demo_path))["demo"] + serve_component(page, demo) + + source = _node_locator(page, "Source").locator(".react-flow__handle-right") + number = _node_locator(page, "Number").locator(".react-flow__handle-right") + transform = _node_locator(page, "Transform").locator(".react-flow__handle-left") + publish = _node_locator(page, "Publish").locator(".react-flow__handle-left") + + def start_drag(handle): + box = handle.bounding_box() + page.mouse.move(box["x"] + box["width"] / 2, box["y"] + box["height"] / 2) + page.mouse.down() + page.mouse.move(box["x"] + box["width"] / 2 + 25, box["y"] + box["height"] / 2) + + start_drag(number) + expect(transform).to_have_class(re.compile("rf-handle-invalid")) + assert "Type mismatch" in transform.get_attribute("data-tooltip") + page.mouse.up() + + start_drag(source) + expect(publish).to_have_attribute("data-tooltip", "Publish requires text from Transform, not Source.") + box = publish.bounding_box() + page.mouse.move(box["x"] + box["width"] / 2, box["y"] + box["height"] / 2) + page.mouse.up() + assert demo._flow.edges == [] + + start_drag(source) + expect(transform).to_have_class(re.compile("rf-handle-valid")) + box = transform.bounding_box() + page.mouse.move(box["x"] + box["width"] / 2, box["y"] + box["height"] / 2) + page.mouse.up() + wait_until(lambda: len(demo._flow.edges) == 1, timeout=8000) + expect(page.get_by_text("Connected: source.text to transform.text")).to_be_visible() + def test_selection_syncs_between_ui_and_python(page): flow = _make_flow() diff --git a/zensical.toml b/zensical.toml index 3b01cfa..ea9d9d8 100644 --- a/zensical.toml +++ b/zensical.toml @@ -18,6 +18,7 @@ nav = [ {"Define Editors" = "how-to/define-editors.md"}, {"Embed Views in Nodes" = "how-to/embed-views-in-nodes.md"}, {"Style Nodes & Edges" = "how-to/style-nodes-edges.md"}, + {"Validate Connections" = "how-to/validate-connections.md"}, {"React to Events" = "how-to/react-to-events.md"}, {"Recover from Rendering Errors" = "how-to/recover-from-errors.md"} ]},