Reference¶

class Edge(param.Parameterized):
    """Base class for object-oriented edges."""

    id = param.String(default="", doc="Unique edge identifier.")
    source = param.String(default="", doc="Source node id.")
    target = param.String(default="", doc="Target node id.")
    label = param.String(default=None, allow_None=True, doc="Display label.")
    type = param.String(default=None, allow_None=True, doc="Edge type.")
    selected = param.Boolean(default=False, doc="Selection state.")
    data = param.Dict(default={}, doc="Custom edge data.")
    style = param.Dict(default=None, allow_None=True, doc="Optional edge style.")
    markerEnd = param.Dict(default=None, allow_None=True, doc="Optional edge end marker.")
    sourceHandle = param.String(default=None, allow_None=True, doc="Optional source handle id.")
    targetHandle = param.String(default=None, allow_None=True, doc="Optional target handle id.")

    @classmethod
    def _data_param_names(cls) -> list[str]:
        return _parameterized_data_param_names(cls, Edge)

    @classmethod
    def _data_schema(cls) -> dict[str, Any]:
        return _parameterized_data_schema(cls, Edge)

    def to_dict(self) -> dict[str, Any]:
        """Convert this edge instance to a ReactFlow-compatible dictionary."""
        data = dict(self.data or {})
        for name in self._data_param_names():
            data[name] = getattr(self, name)
        payload = {
            "id": self.id,
            "source": self.source,
            "target": self.target,
            "label": self.label,
            "type": self.type,
            "selected": self.selected,
            "data": data,
        }
        if self.style is not None:
            payload["style"] = dict(self.style)
        if self.markerEnd is not None:
            payload["markerEnd"] = dict(self.markerEnd)
        if self.sourceHandle is not None:
            payload["sourceHandle"] = self.sourceHandle
        if self.targetHandle is not None:
            payload["targetHandle"] = self.targetHandle
        return payload

    def editor(self, data, schema, *, id, type, on_patch):
        """Optional per-edge editor factory.

        Return ``None`` to fall back to type/default editors.
        """
        return None

    def on_event(self, payload: dict[str, Any], flow: "ReactFlow") -> None:
        """Wildcard event hook for edge-related events."""

    def on_add(self, payload: dict[str, Any], flow: "ReactFlow") -> None:
        """Hook called when this edge is added."""

    def on_delete(self, payload: dict[str, Any], flow: "ReactFlow") -> None:
        """Hook called when this edge is deleted."""

    def on_data_change(self, payload: dict[str, Any], flow: "ReactFlow") -> None:
        """Hook called when this edge's data changes."""

    def on_selection_changed(self, payload: dict[str, Any], flow: "ReactFlow") -> None:
        """Hook called when this edge participates in a selection update."""

    def on_sync(self, payload: dict[str, Any], flow: "ReactFlow") -> None:
        """Hook called when the graph receives a sync payload."""

@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

class Node(param.Parameterized):
    """Base class for object-oriented nodes.

    Subclass this class when you want node instances to keep Python-side state
    and react to graph events directly. Node instances can be passed anywhere
    a node dict/``NodeSpec`` is accepted.

    Subclasses can customize:

    - ``__panel__`` to render node content.
    - ``editor`` to provide a node-specific editor.
    - ``on_event`` (wildcard) and event-specific ``on_*`` hooks.
    """

    id = param.String(default="", doc="Unique node identifier.")
    position = param.Dict(default={"x": 0.0, "y": 0.0}, doc="Node position.")
    type = param.String(default="panel", doc="Node type.")
    label = param.String(default=None, allow_None=True, doc="Display label.")
    data = param.Dict(default={}, doc="Custom node data.")
    selected = param.Boolean(default=False, doc="Selection state.")
    draggable = param.Boolean(default=True, doc="Whether node is draggable.")
    connectable = param.Boolean(default=True, doc="Whether node is connectable.")
    deletable = param.Boolean(default=True, doc="Whether node is deletable.")
    style = param.Dict(default=None, allow_None=True, doc="Optional node style.")
    className = param.String(default=None, allow_None=True, doc="Optional CSS class.")

    @classmethod
    def _data_param_names(cls) -> list[str]:
        return _parameterized_data_param_names(cls, Node)

    @classmethod
    def _data_schema(cls) -> dict[str, Any]:
        return _parameterized_data_schema(cls, Node)

    def to_dict(self) -> dict[str, Any]:
        """Convert this node instance to a ReactFlow-compatible dictionary."""
        data = dict(self.data or {})
        for name in self._data_param_names():
            data[name] = getattr(self, name)
        payload = {
            "id": self.id,
            "position": dict(self.position or {"x": 0.0, "y": 0.0}),
            "type": self.type or "panel",
            "label": self.label,
            "data": data,
            "selected": self.selected,
            "draggable": self.draggable,
            "connectable": self.connectable,
            "deletable": self.deletable,
        }
        if self.style is not None:
            payload["style"] = dict(self.style)
        if self.className is not None:
            payload["className"] = self.className
        view = self.__panel__()
        if view is not None:
            payload["view"] = view
        return payload

    def __panel__(self) -> Any | None:
        """Optional view rendered inside the node."""
        return None

    def editor(self, data, schema, *, id, type, on_patch):
        """Optional per-node editor factory.

        Return ``None`` to fall back to type/default editors.
        """
        return None

    def on_event(self, payload: dict[str, Any], flow: "ReactFlow") -> None:
        """Wildcard event hook for node-related events."""

    def on_add(self, payload: dict[str, Any], flow: "ReactFlow") -> None:
        """Hook called when this node is added."""

    def on_delete(self, payload: dict[str, Any], flow: "ReactFlow") -> None:
        """Hook called when this node is deleted."""

    def on_move(self, payload: dict[str, Any], flow: "ReactFlow") -> None:
        """Hook called when this node moves."""

    def on_click(self, payload: dict[str, Any], flow: "ReactFlow") -> None:
        """Hook called when this node is clicked."""

    def on_data_change(self, payload: dict[str, Any], flow: "ReactFlow") -> None:
        """Hook called when this node's data changes."""

    def on_selection_changed(self, payload: dict[str, Any], flow: "ReactFlow") -> None:
        """Hook called when this node participates in a selection update."""

    def on_sync(self, payload: dict[str, Any], flow: "ReactFlow") -> None:
        """Hook called when the graph receives a sync payload."""