Python Node Development

[ricciV1]

Hello KNIMErs!

I would like to use the ConnectionPortObject to pass my BrowserSession onto another node.
I tried many ways to implement my output port but seem to get nowhere.

As of right now I use this code to implement my output port:

class StarterNode(knext.PythonNode):
def __init__(self):
        self.output_ports = [
            knext.Port(
                type=ConnectionPortObject,
                name="Browser Session",
                description="Browser"
            )
            ]

To pass my data i tried to use

 def execute(self, exec_context):
       
       
        match self.browser_selection.driverSelection:

            case "Chrome":
                driver = chrome_window()
                driver.get(self.browser_configuration.urlSelection)

            case "Edge":
                driver = edge_window()
                driver.get(self.browser_configuration.urlSelection)

            case "Firefox":
                driver = firefox_window()
                driver.get(self.browser_configuration.urlSelection)

        
        
        return knext.ConnectionPortObject

I get the following error message: “Execute failed: type object ‘ConnectionPortObject’ has no attribute ‘id’”
Although i tried to add an ID, I get the same message.

I know I have to pass my “driver” - data, but I cannot find a solution.

Different AI suggested to implement a class like:

class BrowserSessionObject(ConnectionPortObjects)...

… but that got me nowhere as well.

What am I missing here?

Thank you in advance!

[bwilhelm]

Hi @ricciV1,

There isn’t a complete example available for connection port objects, but you can define them similarly to other custom port objects. Extend the ConnectionPortObject class instead of PortObject in your implementation and override the necessary methods.

Useful resources:

• Guide on defining custom port objects
• API documentation on methods to override

Hope this helps!

[ricciV1]

Thank you @bwilhelm for sharing!
It helped me setting up my output-port for my first Node. However, using the same code, I have trouble setting up my input port.
Please find my code below:

Code from my 2nd node:

class MyPortObjectSpec(knext.PortObjectSpec):
    def __init__(self) -> None:
        super().__init__()
        #self._spec_data = spec_data

    def serialize(self) -> dict:
        return

    @classmethod
    def deserialize(cls, data: dict) -> "MyPortObjectSpec":
        cls(data["spec_data"])

class BrowserSessionObject(ConnectionPortObject):
    def __init__(self, spec: MyPortObjectSpec) -> None:
        super().__init__(spec)

    def to_connection_data(self) -> any:
        return super().to_connection_data()
    
    @classmethod
    def from_connection_data(cls, spec: MyPortObjectSpec, data: any) -> ConnectionPortObject:
        return super().from_connection_data(spec, data)

myPortType = knext.port_type(name="Browser Session", object_class=BrowserSessionObject, spec_class=MyPortObjectSpec)
@knext.node(name="Delete Browser Session", node_type=knext.NodeType.OTHER, icon_path="icon.png", category=category)
@knext.input_port(name="Browser Session", description="fhg", port_type=myPortType)
class deleteWebinteracion(knext.PythonNode):
    """
    Description
    """
    
    def configure(self, config_context):
        return 
    
    def execute(self, exec_context):
        return

I am able to connect my ports but after executing my first node, I receive the following error message:

The provided input port type Browser Session must be the same or a sub-type of the node’s input port type Browser Session.

In my first node I pass my “data” like this:

def execute(self, exec_context) -> BrowserSessionObject:
        ...

        myPortSpec = MyPortObjectSpec()
        browserSession = BrowserSessionObject(myPortSpec)
        
        return browserSession

Does anyone know how I have to adjust my code?

[carstenhaubold]

Hi @ricciV1,

The error message says that the type of the input port of the node where you showed the code does not match the type of the output port of your “first” node. What does the @knext.output_port line look like for your first node?

Cheers,
Carsten

[ricciV1]

Hi @carstenhaubold,

my output_port is created the same way as in my 2nd node. So it would be the following:

... (same as in second node -> see above)

myPortType = knext.port_type(name="Browser Session", object_class=BrowserSessionObject, spec_class=MyPortObjectSpec)

@knext.output_port(name="Browser Session", description="The output of your current browser session.", port_type=myPortType)

[carstenhaubold]

Hi @ricciV1,

That’s interesting. Are you registering the myPortType twice? This should only be possible once. You need to use the same registered port type for all ports that shall work with your Browser Session.

[ricciV1]

Hi @carstenhaubold,

by registering twice do you mean if I accidentally used this code twice in the same node:

myPortType = knext.port_type(name="Browser Session", object_class=BrowserSessionObject, spec_class=MyPortObjectSpec)

… or is there some code I need to insert to actually “register” myPortType.

I can also provide the code I have written so far, this may be easier
createBrowserSession.txt (6.1 KB)
deleteBrowserSession.txt (1.7 KB)

[mummec]

Hi, I have the feeling that we running a bit in circles.
Do you have an ad-hoc concrete example on how we hand over this kind of port type?`
Thank you in advance

[mrabdullah]

• Create a Custom Connection Object: You need to define a custom class that inherits from ConnectionPortObject. This class should store your driver instance and implement any required methods, including the get_id() method.

python

Copy code

class BrowserSessionObject(knext.ConnectionPortObject):
    def __init__(self, driver):
        self.driver = driver
    
    def get_id(self):
        return "BrowserSessionObject"

• Modify the execute Method: In your execute method, you need to return an instance of your custom BrowserSessionObject class with the driver as a parameter.

python

Copy code

def execute(self, exec_context):
    match self.browser_selection.driverSelection:
        case "Chrome":
            driver = chrome_window()
            driver.get(self.browser_configuration.urlSelection)

        case "Edge":
            driver = edge_window()
            driver.get(self.browser_configuration.urlSelection)

        case "Firefox":
            driver = firefox_window()
            driver.get(self.browser_configuration.urlSelection)
    
    # Return the custom ConnectionPortObject
    return BrowserSessionObject(driver)

• Ensure Proper Registration: Make sure that your custom BrowserSessionObject is properly registered within KNIME so that the platform recognizes and processes it correctly.

[ricciV1]

@mrabdullah Thank you, but unfortunately this did not help.

[MartinDDDD]

Have not (yet) used Connection Port Objects, but can share what helped me understand better how it all works for custom port objects.

1. Find a KNIME extension that “does something broadly similar” to what you plan to do and that is python based - for your case the KNIME Web Interaction extension ticks that box: KNIME Web Interaction (Labs) – KNIME Community Hub
2. Install the extension and then inspect the python code. For the Web Interactions Extension you can find the node implementations in this folder (starting from your KNIME root folder): knime\plugins\org.knime.python.web_5.3.0.v202407021847\src\main\python\nodes\ ==> interact.py, extract.py invoke.py
   The Port Objects are defined here:knime\plugins\org.knime.python.web_5.3.0.v202407021847\src\main\python\util\common.py
3. study the set up and the flow by looking at a concrete example…

Sneak peak of the connection objects used by the above mentioned extension from common.py:

[ricciV1]

Solved!
After some trying I finally found the issue.

1. Develop your custom port type in a separate file!
   Example:

from typing import Dict
import knime.extension as knext
from knime.extension import ConnectionPortObject
from knime.api.schema import PortObjectSpec


class MyPortObjectSpec(knext.PortObjectSpec):
    def __init__(self) -> None:
        super().__init__()
        # self._spec_data = spec_data

    def serialize(self) -> dict:
        return

    @classmethod
    def deserialize(cls, data: Dict):
        return super().deserialize(data)


class BrowserSessionObject(ConnectionPortObject):
    def __init__(self, spec: MyPortObjectSpec, driver: "driver") -> None:
        self._driver = driver
        super().__init__(spec)

    @property
    def spec(self):
        return super().spec

    @property
    def driver(self):
        return self._driver

    def to_connection_data(self) -> any:
        return {"driver": self._driver}

    @classmethod
    def from_connection_data(
        cls, spec: knext.PortObjectSpec, data: any
    ) -> ConnectionPortObject:
        return cls(spec, data["driver"])


customPortType = knext.port_type(
    name="Browser Session",
    object_class=BrowserSessionObject,
    spec_class=MyPortObjectSpec,
)

2. Import your custom port type to the node.

from util.customPortClass import BrowserSessionObject, MyPortObjectSpec, customPortType

3. Use the port_type like this:

@knext.output_port(
    name="Browser Session",
    description="The output of your current browser session.",
    port_type=customPortType,
)