Python Node: How to handle a credential output port

[Loo]

Hi there

I’m trying to create a Python node with an output port of type knext.PortType.CREDENTIAL - similar to the existing Google Authenticator. This node authenticates with an OAuth service and passes the credentials to other nodes down the line. In order to avoid sending those in clear text, I tried to use a credential port.

However, I’m stuck at specifying the schema within the configure method. I think the knext.CredentialPortObjectSpec object [source] could be what I’m looking for. But the constructor requires a java_callback parameter which I do not know how to handle.

How do I specify the schema within the configure method and the returned data within the execute method for a credential port?

Best
Lo

P.S. Maybe this is not possible with Python, but only with a Java node. It looks like nodes written in Java can use some features which are not (yet) accessible for Python nodes.

[diego_rod_lop]

Hey @Loo

First, thanks for posting in the KNIME forum; I hope it will be the first of multiple.

It might be worth mentioning that a KNIME Authentication extension is already available, which allows you to authenticate via client credentials or interactive authentication.

For example, you can use the OAuth2 Authenticator node, which is very flexible. Indeed, you can use custom endpoints or specific prefilled services.

The output port of the OAuth2 Authenticator node is of type Credential. So, If it is not strictly necessary, I recommend you use this node to pass the access token (credentials) to the downstream nodes you are developing.

You can use a port port_type=knext.PortType.CREDENTIAL"as an input port, and you can access the token in both methods, the configure() and the execute() ones.

Below, I’ve included a dummy code snippet of a node that fetches my GitHub profile information. You need a GitHub account, a Client ID, and a Secret.

@knext.input_port(
    name="Input",
    description="Input port for testing",
    port_type=knext.PortType.CREDENTIAL,
)
@knext.output_port(
    name="Output",
    description="Output port for testing",
    port_type=knext.PortType.TABLE,
)
class CredentialsTest:
    """
    This node provides credentials for testing purposes.
    """

    api_endpoint = knext.StringParameter(
        label="API Endpoint",
        description="Endpoint to interact with the API.",
        default_value="https://api.github.com/user",
    )

    def configure(
        self,
        configuration_context: knext.ConfigurationContext,
        credential_port: knext.CredentialPortObjectSpec,
    ):
        """
        This node provides credentials for testing purposes.
        """
        LOGGER.warning("Configuring credentials test node.")
        LOGGER.warning(f"Port spec: {credential_port}")
        LOGGER.warning(f"Port spec type: {type(credential_port)}")
        LOGGER.warning(f"Port auth: {credential_port.auth_parameters}")
        LOGGER.warning(f"Port auth type: {type(credential_port.auth_parameters)}")
        LOGGER.warning(f"Port schema: {credential_port.auth_schema}")
        LOGGER.warning(f"Port schema type: {type(credential_port.auth_schema)}")
        pass

    def execute(
        self, exec_context: knext.ExecutionContext, credential: knext.PortObject
    ):
        """
        This node provides credentials for testing purposes.
        """
        api_url = self.api_endpoint
        access_token = credential.spec.auth_parameters
        LOGGER.warning(f"Access token: {access_token}")
        LOGGER.warning(f"inside the specs {dir(credential.spec)}")

        headers = {
            "Authorization": f"Bearer {access_token}",
            "Accept": "application/vnd.github.v3+json",
        }

        response = requests.get(api_url, headers=headers)

        profile_data = response.json()

        # Create a pandas DataFrame to store the profile data
        data = {
            "Username": [profile_data["login"]],
            "Name": [profile_data.get("name", "No name available")],
            "Public Repos": [profile_data["public_repos"]],
            "Followers": [profile_data["followers"]],
            "Following": [profile_data["following"]],
            "Bio": [profile_data.get("bio", "No bio available")],
        }

        # Convert the dictionary to a DataFrame
        df = pd.DataFrame(data)

        # Display the DataFrame
        return knext.Table.from_pandas(df)

Below is the screenshot of the workflow that is using the node:

I don’t know which kind of OAuth service are you trying to authenticate and if you need to refresh the token, in case we can check how to do it.

BR,