Port Objects

Port Object Specs

class knime.extension.PortObjectSpec

Base protocol for port object specs.

A PortObjectSpec must support conversion from/to a dictionary which is then encoded as JSON and sent to/from KNIME.

classmethod deserialize(data: Dict, java_callback = None)

Deserialize the port object.

Parameters

• data : Dict — The data dict created by serialize
• java_callback : any — A callback handler that allows to make callbacks to Java. Was introduced in 5.3 and can be omitted in derived classes.

Returns

PortObjectSpec — The deserialized PortObjectSpec

method serialize() → dict

class knime.extension.BinaryPortObjectSpec

Port object spec for simple binary port objects.

BinaryPortObjectSpecs have an ID that is used to ensure that only ports with equal ID can be connected.

method __init__(id: str) → None

Create a BinaryPortObjectSpec.

Parameters

• id : str — The id of this binary port.

classmethod deserialize(data)

property id → str

method serialize() → dict

class knime.extension.ImagePortObjectSpec

Port object spec for image port objects.

ImagePortObjectSpec objects require the format specified via knext.ImageFormat.PNG or knext.ImageFormat.SVG.

method __init__(format: Union[str, Enum]) → None

Create an ImagePortObjectSpec

Parameters

• format : Union[str, Enum] — The format of the image expected to pass through the port.

classmethod deserialize(data)

property format → str

method serialize() → dict

class knime.extension.ImageFormat

The image formats available for image ports.

classmethod available_options()

attribute PNG = 'png'

The PNG format.

attribute SVG = 'svg'

The SVG format.

Custom Port Object Types

class knime.extension.PortObject

Base class for custom port objects. They must have a corresponding PortObjectSpec and support serialization from and to bytes.

Examples

>>> class MyPortObjectSpec(knext.PortObjectSpec):
...     def __init__(self, my_info: str) -> None:
...         self._my_info = my_info
...
...     def serialize(self) -> dict:
...         return {
...             "my_info": self._my_info,
...         }
...
...     @classmethod
...     def deserialize(cls, data: dict) -> "MyPortObjectSpec":
...         return cls(data["my_info"])
...
...     @property
...     def my_info(self) -> str:
...         return self._my_info
...
... class MyPortObject(knext.PortObject):
...     def __init__(self, spec: MyPortObjectSpec, my_data: str) -> None:
...         super().__init__(spec)
...         self._my_data = my_data
...
...     def serialize(self) -> bytes:
...         return self._my_data.encode()
...
...     @property
...     def spec(self) -> MyPortObjectSpec:
...         return super().spec
...
...     @classmethod
...     def deserialize(cls, spec: MyPortObjectSpec, storage: bytes) -> "MyPortObject":
...         return cls(spec, storage.decode())
...
...     @property
...     def my_data(self) -> str:
...         return self._my_data
...
...
... my_port_type = knext.port_type("My Port Type", MyPortObject, MyPortObjectSpec)
...
... @knext.node(
...     name="My PortObject Creator",
...     node_type=knext.NodeType.SOURCE,
...     icon_path="icon.png",
...     category=node_category,
... )
... @knext.output_port("MyPortType output", "MyPortType output", my_port_type)
... class NodeWithMyOutputPort(knext.PythonNode):
...     value = knext.StringParameter(
...         "port object value",
...         "Value that will be put in the port object",
...         "Foo",
...     )
...
...     def configure(self, config_context: knext.ConfigurationContext):
...         return MyPortObjectSpec(f"port object will contain value {self.value}")
...
...     def execute(self, exec_context: knext.ExecutionContext):
...         return MyPortObject(MyPortObjectSpec(f"port object will contain value {self.value}"), self.value)
...
...
... @knext.node(
...     name="My PortObject Applier",
...     node_type=knext.NodeType.MANIPULATOR,
...     icon_path="icon.png",
...     category=node_category,
... )
... @knext.input_port("MyPortType input", "MyPortType input", my_port_type)
... @knext.input_table("Input table", "Table to apply some action to.")
... @knext.output_table("Output table", "Table with the applied action.")
... class NodeWithTestInputPort(knext.PythonNode):
...     '''
...     This node adds a column containing some info from the port object
...     '''
...     def configure(
...         self,
...         config_context: knext.ConfigurationContext,
...         spec: MyPortObjectSpec,
...         schema: knext.Schema,
...     ):
...         return schema.append(knext.Column(knext.string(), "NewCol"))
...
...     def execute(
...         self,
...         exec_context: knext.ExecutionContext,
...         port_object: MyPortObject,
...         table: knext.Table,
...     ):
...         df = table.to_pandas()
...         df["NewCol"] = [port_object.my_data] * len(df)
...
...         return knext.Table.from_pandas(df)

method __init__(spec: PortObjectSpec) → None

classmethod deserialize(spec: PortObjectSpec, storage: bytes) → PortObject

Creates the port object from its spec and storage.

method serialize() → bytes

Serialize the object to bytes.

property spec → PortObjectSpec

Provides access to the spec of the PortObject.

class knime.extension.ConnectionPortObject

Connection port objects are a special type of port objects which support dealing with non-serializable objects such as database connections or web sessions.

Connection port objects are passed downstream by ensuring that the same Python process is used to execute subsequent nodes. ConnectionPortObjects must provide the data in the to_connection_data and create new instances from the same data in from_connection_data. A reference to the data Python object is maintained and handed to downstream nodes. So the data does not need to be serializable/picklable.

method __init__(spec: PortObjectSpec) → None

classmethod deserialize(spec, storage)

classmethod from_connection_data(spec: PortObjectSpec, data: Any) → ConnectionPortObject

Construct a ConnectionPortObject from spec and data. The data is the data that has been returned by the to_connection_data method of the ConnectionPortObject by the upstream node.

The data should not be tempered with, as it is a Python object that is handed to all nodes using this ConnectionPortObject.

method serialize()

property spec → PortObjectSpec

Provides access to the spec of the PortObject.

method to_connection_data() → Any

Provide the data that makes up this ConnectionPortObject such that it can be used by downstream nodes in the from_connection_data method.