node

@dataclass
class Node:
    """A node within a container (graph or group)."""

    plugin_ref: PluginRef
    name: str
    label: Optional[str] = None
    static_values: Dict[str, Any] = field(default_factory=dict)
    input_connections: Dict[str, Union[NodeConnection,
        List[NodeConnection], None]] = field(
            default_factory=dict
    )

    ui: Optional[ui_utils.UiNodeMetadata] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        plugin_desc = self.get_plugin_desc()
        self._multi_input_names = []

        for plg_in in plugin_desc.inputs:
            name = plg_in.name
            self.static_values.setdefault(name, get_input_default(plg_in))
            if is_multi_connection_input(plg_in):
                self._multi_input_names.append(name)
                self.input_connections.setdefault(name, [])
            else:
                self.input_connections.setdefault(name, None)

        self._outputs = dict.fromkeys(
            [plg_out.name for plg_out in plugin_desc.outputs]
        )

        # Mandatory revert output slot for revert features.
        self._outputs["revert"] = None

    def get_plugin_desc(self) -> WorkflowNode:
        plugins = PluginRegistry().plugin_list()
        for name, plugin_desc in plugins.items():
            if (
                name == self.plugin_ref.name
                # TODO: check backward-compatibility here.
                and plugin_desc.version == self.plugin_ref.version
            ):
                return plugin_desc

        raise ValueError(f"Unknown plugin: {self.plugin_ref}")

    def __str__(self) -> str:
        return f"<Node {self.label} ({self.plugin_ref})>"

    def __setitem__(self, input_name: str, static_value: object):
        """Set node input static value."""
        if input_name not in self.inputs:
            raise KeyError(
                f"Invalid input name {input_name} for node {self}. "
                f"Available inputs are: {sorted(self.inputs)}"
            )

        # TODO: bring ¸input type validation against plugin_desc here.
        self.static_values[input_name] = static_value

    def __getitem__(self, input_name: str) -> object:
        """Get node input (static value or connection)."""
        if input_name not in self.inputs:
            raise KeyError(
                f"Invalid input name {input_name} for node {self}. "
                f"Available inputs are: {sorted(self.inputs)}"
            )

        return (
            # input comes from an inward node connection
            self.input_connections.get(input_name)
            # or is it set as static value ?
            or self.static_values[input_name]
        )

    @property
    def display_name(self) -> str:
        return self.label or self.name

    @property
    def inputs(self) -> Tuple:
        return tuple(self.input_connections.keys())

    @property
    def outputs(self) -> Tuple[str]:
        return tuple(self._outputs.keys())

    @property
    def node_type(self) -> str:
        return self.plugin_ref.name

    def _set_input(
        self, origin_node_name: str, origin_output_name: str, input_name: str
    ) -> None:
        if input_name not in self.inputs:
            raise ValueError(
                f"Cannot connect {origin_node_name}.{origin_output_name} "
                f"to {self}.{input_name}. Provided input does "
                f"not exist in {self}."
            )

        # TODO: bring ¸input type validation against node output type here.
        node_connection = NodeConnection(
            origin_node_name=origin_node_name,
            origin_node_output=origin_output_name,
        )

        if input_name in self._multi_input_names:
            self.input_connections[input_name].append(node_connection)
        else:
            self.input_connections[input_name] = node_connection

    def connect(
        self,
        output_name: str,
        destination_node: "Node",
        destination_input_name: str,
    ):
        """Connect specific output to another node input."""
        if output_name not in self._outputs:
            raise ValueError(
                f"Provided output {output_name} does not exist in {self}. "
                f"Available outputs are {self._outputs.keys()}"
            )

        if destination_node is self:
            raise ValueError(f"Cannot connect node to itself {self}.")

        # Actual implementation is to register
        # current node as destination node input.
        destination_node._set_input(
            self.name, output_name, destination_input_name
        )

    def disconnect_input(self,
            input_name: str,
            input_idx: int = -1,
        ):
        """Reset connection of incoming input."""
        if (
            input_name not in self.inputs
            or not self.input_connections[input_name]
        ):
            return

        if input_name not in self._multi_input_names:
            self.input_connections[input_name] = None
            return

        try:
            self.input_connections[input_name].pop(input_idx)
        except IndexError:
            return

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "Node":
        data = serialization.deserialize_from_primitive(data)

        # Backward compatible with previous serialization method.
        # TODO remove legacy serialization format support in a future version.
        if isinstance(data, dict):
            logger.warning("Node.from_dict: using legacy serialization format")
            input_connections = {}
            for key, value in data.get("input_connections", {}).items():
                if isinstance(value, list):
                    value = [NodeConnection(**val) for val in value]
                elif value:
                    value = NodeConnection(**value)

                input_connections[key] = value

            return cls(
                name=data["name"],
                plugin_ref=PluginRef(**data["plugin_ref"]),
                label=data.get("label"),
                static_values=data.get("static_values", {}),
                input_connections=input_connections,
                ui=data.get("ui"),
            )

        if not isinstance(data, cls):
            raise TypeError(
                f"Node.from_dict expected to deserialize a {cls.__name__} "
                f"instance, but got {type(value).__name__!r} instead."
            )

        return data

@dataclass
class NodeConnection:
    """A connection to a specific node."""

    origin_node_name: str
    origin_node_output: str

@dataclass
class PluginRef:
    """A reference to a Plugin (name + version)."""

    name: str
    version: str

    def __str__(self) -> str:
        return f"{self.name}@{self.version}"