Reference¶

@dataclass
class EdgeSpec:
    """Builder for edge dictionaries with validation and type safety.

    This helper class simplifies edge creation by providing a structured
    interface with sensible defaults. It ensures all required fields are
    present and provides convenient conversion to/from dictionaries.

    Parameters
    ----------
    id : str
        Unique identifier for the edge. Must be unique within the graph.
    source : str
        ID of the source node where the edge originates.
    target : str
        ID of the target node where the edge terminates.
    label : str, optional
        Display label shown on the edge. If ``None``, no label is displayed.
    type : str, optional
        Edge type identifier. Reference a custom type defined in
        ``ReactFlow.edge_types`` for schema validation and custom rendering.
    selected : bool, default False
        Whether the edge is currently selected in the UI.
    data : dict, optional
        Custom data dictionary for the edge. Defaults to ``{}`` if not provided.
        This is where you store edge-specific properties that match the schema.
    style : dict, optional
        CSS style dictionary applied to the edge line. Example:
        ``{"stroke": "#ff0000", "strokeWidth": 3}``
    markerEnd : dict, optional
        Arrow marker configuration for the edge end. Example:
        ``{"type": "arrow", "color": "#000000"}``
    sourceHandle : str, optional
        ID of the specific handle on the source node where the edge originates.
        Use this when the source node has multiple output handles defined.
    targetHandle : str, optional
        ID of the specific handle on the target node where the edge terminates.
        Use this when the target node has multiple input handles defined.

    Methods
    -------
    to_dict()
        Convert to a dictionary for use with ReactFlow.
    from_dict(payload)
        Create an EdgeSpec from a dictionary.

    Examples
    --------
    Create a basic edge:

    >>> from panel_reactflow import EdgeSpec
    >>> edge = EdgeSpec(
    ...     id="edge1",
    ...     source="node1",
    ...     target="node2"
    ... )
    >>> edge_dict = edge.to_dict()

    Create an edge with styling:

    >>> edge = EdgeSpec(
    ...     id="edge2",
    ...     source="node2",
    ...     target="node3",
    ...     label="Connection",
    ...     style={"stroke": "#1976d2", "strokeWidth": 2},
    ...     markerEnd={"type": "arrowclosed", "color": "#1976d2"}
    ... )

    Create a typed edge with data:

    >>> edge = EdgeSpec(
    ...     id="weighted_edge",
    ...     source="n1",
    ...     target="n2",
    ...     type="weighted",
    ...     label="0.75",
    ...     data={"weight": 0.75, "confidence": 0.9}
    ... )

    Create an edge with specific handles:

    >>> edge = EdgeSpec(
    ...     id="handle_edge",
    ...     source="producer",
    ...     target="consumer",
    ...     sourceHandle="result",
    ...     targetHandle="mode"
    ... )

    Add to a ReactFlow graph:

    >>> from panel_reactflow import ReactFlow
    >>> flow = ReactFlow()
    >>> flow.add_edge(edge)
    """

    id: str
    source: str
    target: str
    label: str | None = None
    type: str | None = None
    selected: bool = False
    data: dict[str, Any] | None = None
    style: dict[str, Any] | None = None
    markerEnd: dict[str, Any] | None = None
    sourceHandle: str | None = None
    targetHandle: str | None = None

    def __post_init__(self) -> None:
        if self.data is None:
            self.data = {}

    def to_dict(self) -> dict[str, Any]:
        """Convert the EdgeSpec to a dictionary.

        Returns
        -------
        dict
            Dictionary representation suitable for ReactFlow.
        """
        payload = {
            "id": self.id,
            "source": self.source,
            "target": self.target,
            "label": self.label,
            "type": self.type,
            "selected": self.selected,
            "data": self.data,
        }
        if self.style is not None:
            payload["style"] = self.style
        if self.markerEnd is not None:
            payload["markerEnd"] = self.markerEnd
        if self.sourceHandle is not None:
            payload["sourceHandle"] = self.sourceHandle
        if self.targetHandle is not None:
            payload["targetHandle"] = self.targetHandle
        return payload

    @classmethod
    def from_dict(cls, payload: dict[str, Any]) -> "EdgeSpec":
        """Create an EdgeSpec from a dictionary.

        Parameters
        ----------
        payload : dict
            Dictionary containing edge properties.

        Returns
        -------
        EdgeSpec
            A new EdgeSpec instance.
        """
        return cls(**payload)

@dataclass
class EdgeType:
    """Define a custom edge type with schema for edge properties.

    Edge types allow you to define reusable edge templates with specific data
    schemas for validation and editor generation. Use this when your edges
    have custom properties beyond the basic source/target relationship.

    Parameters
    ----------
    type : str
        Unique identifier for this edge type. Used to reference this type
        when creating edges.
    label : str, optional
        Human-readable display name for this edge type. If not provided,
        the ``type`` value is used.
    schema : dict or type, optional
        Data schema for edge validation and editor generation. Accepts the
        same formats as :class:`NodeType`:

        - A JSON Schema dictionary
        - A ``param.Parameterized`` subclass
        - A Pydantic ``BaseModel`` subclass
        - A :class:`SchemaSource` wrapper for explicit schema types

    Methods
    -------
    to_dict()
        Convert this edge type to a JSON-serializable dictionary.

    Examples
    --------
    Define an edge type with properties:

    >>> from panel_reactflow import EdgeType
    >>> weighted_edge = EdgeType(
    ...     type="weighted",
    ...     label="Weighted Connection",
    ...     schema={
    ...         "type": "object",
    ...         "properties": {
    ...             "weight": {"type": "number", "minimum": 0, "maximum": 1},
    ...             "label": {"type": "string"}
    ...         }
    ...     }
    ... )

    Use the edge type in a ReactFlow graph:

    >>> from panel_reactflow import ReactFlow, EdgeSpec
    >>> flow = ReactFlow(edge_types={"weighted": weighted_edge})
    >>> flow.add_edge(EdgeSpec(
    ...     id="e1",
    ...     source="n1",
    ...     target="n2",
    ...     type="weighted",
    ...     data={"weight": 0.75, "label": "strong"}
    ... ))
    """

    type: str
    label: str | None = None
    schema: Any = None

    def to_dict(self) -> dict[str, Any]:
        """Convert the edge type to a JSON-serializable dictionary.

        Returns
        -------
        dict
            Dictionary representation with normalized schema.
        """
        return {
            "type": self.type,
            "label": self.label,
            "schema": _normalize_schema(self.schema),
        }

class Editor(Viewer):
    """Base class for custom node and edge editors.

    The Editor class provides a standardized interface for creating custom
    property editors for nodes and edges. All editors receive a unified
    signature and can report data changes back to the graph through a
    callback mechanism.

    All editor implementations (whether classes or functions) receive this
    unified signature::

        editor(data, schema, *, id, type, on_patch) -> Viewable

    Parameters
    ----------
    data : dict
        Current node or edge data dictionary. This contains all the custom
        properties stored in the node/edge.
    schema : dict or None
        Normalized JSON Schema for the node/edge type, or ``None`` if no
        schema is defined. Use this to drive form generation or validation.
    id : str
        Unique identifier of the node or edge being edited.
    type : str
        Type name of the node or edge being edited.
    on_patch : callable
        Callback function ``on_patch(patch_dict)`` to report data changes
        back to the graph. Call this with a dictionary of updated properties
        when the user modifies data.

    Examples
    --------
    Create a custom editor class:

    >>> import panel as pn
    >>> from panel_reactflow import Editor
    >>>
    >>> class ColorEditor(Editor):
    ...     def __init__(self, data=None, schema=None, **kwargs):
    ...         super().__init__(data, schema, **kwargs)
    ...         self.color_picker = pn.widgets.ColorPicker(
    ...             name="Node Color",
    ...             value=self._data.get("color", "#000000")
    ...         )
    ...         self.color_picker.param.watch(self._on_change, "value")
    ...
    ...     def _on_change(self, event):
    ...         if self._on_patch:
    ...             self._on_patch({"color": event.new})
    ...
    ...     def __panel__(self):
    ...         return self.color_picker

    Use the custom editor:

    >>> from panel_reactflow import ReactFlow
    >>> flow = ReactFlow(
    ...     node_editors={"panel": ColorEditor}
    ... )

    Create an editor as a simple function:

    >>> def simple_editor(data, schema, *, id, type, on_patch):
    ...     widget = pn.widgets.TextInput(
    ...         name="Label",
    ...         value=data.get("label", "")
    ...     )
    ...     widget.param.watch(
    ...         lambda e: on_patch({"label": e.new}),
    ...         "value"
    ...     )
    ...     return widget
    >>>
    >>> flow = ReactFlow(default_node_editor=simple_editor)
    """

    _data = param.Dict(default={}, doc="Node or edge data.")
    _schema = param.Dict(default=None, allow_None=True, doc="JSON Schema for data.")
    _node_id = param.String(default="", doc="Node or edge ID.")
    _node_type = param.String(default="", doc="Node or edge type.")
    _on_patch = param.Callable(default=None, allow_None=True, doc="Callback to report data changes.")

    def __init__(self, data=None, schema=None, *, id="", type="", on_patch=None, **kwargs):
        super().__init__(
            _data=data if data is not None else {},
            _schema=schema,
            _node_id=id,
            _node_type=type or "",
            _on_patch=on_patch,
            **kwargs,
        )

class JsonEditor(Editor):
    """Simple JSON editor for node and edge data.

    This editor provides a raw JSON editing interface using Panel's
    JSONEditor widget. It's useful for debugging or when you want full
    control over the data structure without schema-driven forms.

    The editor automatically syncs changes back to the graph when the
    user modifies the JSON content.

    Parameters
    ----------
    data : dict, optional
        Initial node or edge data dictionary. Defaults to ``{}``.
    schema : dict, optional
        JSON Schema (not used by this editor but part of the standard
        interface). This editor ignores the schema and allows free-form
        JSON editing.
    **kwargs
        Additional keyword arguments passed to the :class:`Editor` base class.

    Examples
    --------
    Use as default editor for all nodes:

    >>> from panel_reactflow import ReactFlow, JsonEditor
    >>> flow = ReactFlow(default_node_editor=JsonEditor)

    Use for specific node types:

    >>> flow = ReactFlow(
    ...     node_editors={"custom": JsonEditor}
    ... )

    The editor will display a JSON editor interface where users can
    directly edit the node's data dictionary.
    """

    def __init__(self, data=None, schema=None, **kwargs):
        super().__init__(data, schema, **kwargs)
        self._editor = JSONEditor(value=self._data)
        self._editor.param.watch(self._on_json_change, "value")

    def _on_json_change(self, event: param.parameterized.Event) -> None:
        if self._on_patch is not None and event.new != self._data:
            self._data = event.new
            self._on_patch(event.new)

    def __panel__(self):
        return self._editor

@dataclass
class NodeSpec:
    """Builder for node dictionaries with validation and type safety.

    This helper class simplifies node creation by providing a structured
    interface with sensible defaults. It ensures all required fields are
    present and provides convenient conversion to/from dictionaries.

    Parameters
    ----------
    id : str
        Unique identifier for the node. Must be unique within the graph.
    position : dict, optional
        Node position with ``x`` and ``y`` coordinates. Defaults to
        ``{"x": 0.0, "y": 0.0}`` if not provided.
    type : str, default "panel"
        Node type identifier. Use ``"panel"`` for basic nodes or reference
        a custom type defined in ``ReactFlow.node_types``.
    label : str, optional
        Display label shown on the node. If ``None``, no label is displayed.
    data : dict, optional
        Custom data dictionary for the node. Defaults to ``{}`` if not provided.
        This is where you store node-specific properties that match the schema.
    selected : bool, default False
        Whether the node is currently selected in the UI.
    draggable : bool, default True
        Whether the node can be dragged by users.
    connectable : bool, default True
        Whether edges can be connected to/from this node.
    deletable : bool, default True
        Whether the node can be deleted by users.
    style : dict, optional
        CSS style dictionary applied to the node. Example:
        ``{"backgroundColor": "#ff0000", "border": "2px solid black"}``
    className : str, optional
        CSS class name applied to the node for custom styling.
    view : Panel viewable, optional
        Optional Panel viewable (widget, pane, layout) to render inside
        the node. The view will be displayed as the node's content.

    Methods
    -------
    to_dict()
        Convert to a dictionary for use with ReactFlow.
    from_dict(payload)
        Create a NodeSpec from a dictionary.

    Examples
    --------
    Create a basic node:

    >>> from panel_reactflow import NodeSpec
    >>> node = NodeSpec(
    ...     id="node1",
    ...     position={"x": 100, "y": 50},
    ...     label="Start Node"
    ... )
    >>> node_dict = node.to_dict()

    Create a node with custom styling:

    >>> node = NodeSpec(
    ...     id="node2",
    ...     position={"x": 200, "y": 100},
    ...     label="Process",
    ...     style={"backgroundColor": "#e3f2fd", "border": "2px solid #1976d2"},
    ...     className="custom-node"
    ... )

    Create a node with data:

    >>> node = NodeSpec(
    ...     id="transform1",
    ...     type="transform",
    ...     position={"x": 300, "y": 150},
    ...     label="Data Transform",
    ...     data={"operation": "filter", "threshold": 0.5}
    ... )

    Create a node with an embedded view:

    >>> import panel as pn
    >>> node = NodeSpec(
    ...     id="plot1",
    ...     position={"x": 400, "y": 200},
    ...     label="Data Plot",
    ...     view=pn.pane.Markdown("# Hello World")
    ... )

    Add to a ReactFlow graph:

    >>> from panel_reactflow import ReactFlow
    >>> flow = ReactFlow()
    >>> flow.add_node(node)
    """

    id: str
    position: dict[str, float] | dict[str, Any] = None
    type: str = "panel"
    label: str | None = None
    data: dict[str, Any] | None = None
    selected: bool = False
    draggable: bool = True
    connectable: bool = True
    deletable: bool = True
    style: dict[str, Any] | None = None
    className: str | None = None
    view: Any | None = None

    def __post_init__(self) -> None:
        if self.position is None:
            self.position = {"x": 0.0, "y": 0.0}
        if self.data is None:
            self.data = {}

    def to_dict(self) -> dict[str, Any]:
        """Convert the NodeSpec to a dictionary.

        Returns
        -------
        dict
            Dictionary representation suitable for ReactFlow.
        """
        payload = {
            "id": self.id,
            "position": self.position,
            "type": self.type,
            "label": self.label,
            "data": self.data,
            "selected": self.selected,
            "draggable": self.draggable,
            "connectable": self.connectable,
            "deletable": self.deletable,
        }
        if self.style is not None:
            payload["style"] = self.style
        if self.className is not None:
            payload["className"] = self.className
        if self.view is not None:
            payload["view"] = self.view
        return payload

    @classmethod
    def from_dict(cls, payload: dict[str, Any]) -> "NodeSpec":
        """Create a NodeSpec from a dictionary.

        Parameters
        ----------
        payload : dict
            Dictionary containing node properties.

        Returns
        -------
        NodeSpec
            A new NodeSpec instance.
        """
        return cls(**payload)

@dataclass
class NodeType:
    """Define a custom node type with schema and port configuration.

    Node types allow you to define reusable node templates with specific data
    schemas, input/output ports, and display policies. When nodes are created
    with this type, they automatically get schema validation and appropriate
    editors.

    Parameters
    ----------
    type : str
        Unique identifier for this node type. Used to reference this type
        when creating nodes.
    label : str, optional
        Human-readable display name for this node type. If not provided,
        the ``type`` value is used.
    schema : dict or type, optional
        Data schema for node validation and editor generation. Accepts:

        - A JSON Schema dictionary
        - A ``param.Parameterized`` subclass
        - A Pydantic ``BaseModel`` subclass
        - A :class:`SchemaSource` wrapper for explicit schema types

        The schema is normalized to JSON Schema format internally.
    inputs : list of str, optional
        List of input port names. If provided, these ports will be rendered
        on the node for incoming connections.
    outputs : list of str, optional
        List of output port names. If provided, these ports will be rendered
        on the node for outgoing connections.
    pane_policy : str, default "single"
        Display policy for Panel viewables inside nodes.

    Methods
    -------
    to_dict()
        Convert this node type to a JSON-serializable dictionary.

    Examples
    --------
    Define a simple node type with a JSON Schema:

    >>> from panel_reactflow import NodeType
    >>> transform_type = NodeType(
    ...     type="transform",
    ...     label="Data Transform",
    ...     schema={
    ...         "type": "object",
    ...         "properties": {
    ...             "operation": {"type": "string", "enum": ["filter", "map", "reduce"]},
    ...             "parameter": {"type": "number"}
    ...         }
    ...     },
    ...     inputs=["input"],
    ...     outputs=["output"]
    ... )

    Define a node type with a Param class:

    >>> import param
    >>> class TransformParams(param.Parameterized):
    ...     operation = param.Selector(default="filter", objects=["filter", "map", "reduce"])
    ...     parameter = param.Number(default=1.0)
    >>> transform_type = NodeType(
    ...     type="transform",
    ...     label="Data Transform",
    ...     schema=TransformParams,
    ...     inputs=["input"],
    ...     outputs=["output"]
    ... )

    Use the node type in a ReactFlow graph:

    >>> from panel_reactflow import ReactFlow, NodeSpec
    >>> flow = ReactFlow(node_types={"transform": transform_type})
    >>> flow.add_node(NodeSpec(
    ...     id="t1",
    ...     type="transform",
    ...     position={"x": 100, "y": 100},
    ...     data={"operation": "filter", "parameter": 2.5}
    ... ))
    """

    type: str
    label: str | None = None
    schema: Any = None
    inputs: list[str] | None = None
    outputs: list[str] | None = None
    pane_policy: str = "single"

    def to_dict(self) -> dict[str, Any]:
        """Convert the node type to a JSON-serializable dictionary.

        Returns
        -------
        dict
            Dictionary representation with normalized schema.
        """
        return {
            "type": self.type,
            "label": self.label,
            "schema": _normalize_schema(self.schema),
            "inputs": self.inputs,
            "outputs": self.outputs,
            "pane_policy": self.pane_policy,
        }