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,
