Feature/tp 3065 add ability and sample to connect with jwt to timbr python http connector with backward compatibility

.github/workflows/install-dependencies-and-run-tests.yml

@@ -35,12 +35,18 @@ jobs:
           ONTOLOGY: ${{ secrets.ONTOLOGY }}
           TOKEN: ${{ secrets.TOKEN }}
           QUERY: ${{ secrets.QUERY }}
-          DATASTORE: ${{ secrets.DATASTORE }}
+          DATASOURCE: ${{ secrets.DATASOURCE }}
           HOSTNAME: ${{ secrets.HOSTNAME }}
           PORT: ${{ secrets.PORT }}
           ENABLED_SSL: ${{ secrets.ENABLED_SSL }}
           VERIFY_SSL: ${{ secrets.VERIFY_SSL }}
           NESTED: ${{ secrets.NESTED }}
           ENABLE_IPV6: ${{ secrets.ENABLE_IPV6 }}
+          JWT_TENANT_ID: ${{ secrets.JWT_TENANT_ID }}
+          JWT_CLIENT_ID: ${{ secrets.JWT_CLIENT_ID }}
+          JWT_USERNAME: ${{ secrets.JWT_USERNAME }}
+          JWT_PASSWORD: ${{ secrets.JWT_PASSWORD }}
+          JWT_SCOPE: ${{ secrets.JWT_SCOPE }}
+          JWT_SECRET: ${{ secrets.JWT_SECRET }}
       run: |
         pytest

examples/example.py

@@ -16,14 +16,16 @@
   enable_IPv6 = <True/False>,
 )
 
-  # url    - Required - String - The IP / Hostname of the Timbr platform.
-  # ontology    - Required - String - The ontology / knowledge graph to connect to.
-  # token       - Required - String - Timbr token value.
-  # query       - Required - String - The query that you want to execute.
-  # datasource  - Optional - String - Add the specific datasource name that you want to query from, the default value is the current active datasource of your ontology.
-  # nested      - Optional - String - Change to 'true' if nested flag needs to be enabled. make sure this flag contains string value not bool value.
-  # verify_ssl  - Optional - Boolean - Verifying the target server's SSL Certificate, use False to disable this process.
-  # enable_IPv6 - Optional - Boolean - Change to 'true' if you are using IPv6 connection.
+  # url           - Required - String - The IP / Hostname of the Timbr platform.
+  # ontology      - Required - String - The ontology / knowledge graph to connect to.
+  # token         - Required - String - Timbr token value or JWT token value. Note: If you are using JWT token, you need to set the is_jwt parameter to True.
+  # query         - Required - String - The query that you want to execute.
+  # datasource    - Optional - String - Add the specific datasource name that you want to query from, the default value is the current active datasource of your ontology.
+  # nested        - Optional - String - Change to 'true' if nested flag needs to be enabled. make sure this flag contains string value not bool value.
+  # verify_ssl    - Optional - Boolean - Verifying the target server's SSL Certificate, use False to disable this process.
+  # enable_IPv6   - Optional - Boolean - Change to 'true' if you are using IPv6 connection.
+  # is_jwt        - Optional - Boolean - Set to True if you are using JWT token, otherwise set to False.
+  # jwt_tenant_id - Optional - String - The tenant ID for JWT authentication
 
 # HTTP example
 response = timbr.run_query(
@@ -39,6 +41,24 @@
 
 print(response)
 
+
+# Example for JWT token usage
+response = timbr.run_query(
+  url = "https://mytimbrenv.com:443",
+  ontology = "my_ontology",
+  token = "my_jwt_token",
+  query = "SELECT * FROM timbr.sys_concepts",
+  datasource = "my_datasource",
+  nested = "false",
+  verify_ssl = True,
+  enable_IPv6 = False,
+  is_jwt = True,
+  jwt_tenant_id = "my_tenant_id",
+)
+
+print(response)
+
+
 # HTTPS example
 response = timbr.simpleQueryExecution(
   url = "https://mytimbrenv.com:443",

pytimbr_api/timbr_http_connector.py

@@ -33,6 +33,8 @@ def run_query(
     nested: str = 'false',
     verify_ssl: bool = True,
     enable_IPv6: bool = False,
+    is_jwt: bool = False,
+    jwt_tenant_id: str = None,
 ):
     datasource_addition = ''
     if datasource:
@@ -44,11 +46,17 @@ def run_query(
 
     headers = {
       'Content-Type': 'application/text',
-      'x-api-key': token,
       'nested': nested,
       'Connection': 'close',
     }
-
+
+    if is_jwt:
+      headers['x-jwt-token'] = token
+      if jwt_tenant_id:
+        headers['x-jwt-tenant-id'] = jwt_tenant_id
+    else:
+      headers['x-api-key'] = token
+
     requests.packages.urllib3.util.connection.HAS_IPV6 = enable_IPv6
     response = requests.post(
       f'{base_url}timbr/openapi/ontology/{ontology}/query{datasource_addition}',

test/conftest.py

@@ -31,4 +31,10 @@ def test_config():
     "verify_ssl": convert_env_to_bool(os.getenv("VERIFY_SSL")),
     "nested": os.getenv("NESTED"),
     "enableIPv6": convert_env_to_bool(os.getenv("ENABLE_IPV6")),
+    "jwt_tenant_id": os.getenv("JWT_TENANT_ID"),
+    "jwt_client_id": os.getenv("JWT_CLIENT_ID"),
+    "jwt_username": os.getenv("JWT_USERNAME"),
+    "jwt_password": os.getenv("JWT_PASSWORD"),
+    "jwt_scope": os.getenv("JWT_SCOPE"),
+    "jwt_secret": os.getenv("JWT_SECRET"),
   }

test/test_jwt_token.py

@@ -0,0 +1,47 @@
+import requests
+from pytimbr_api.timbr_http_connector import run_query
+
+def test_query_using_jwt(test_config):
+    # Azure AD token endpoint URL
+    token_url = f'https://login.microsoftonline.com/{test_config["jwt_tenant_id"]}/oauth2/v2.0/token'
+
+    # Request payload for token exchange
+    payload = {
+        'client_id': test_config["jwt_client_id"],
+        'client_secret': test_config["jwt_secret"],
+        'scope': test_config["jwt_scope"],
+        'username': test_config["jwt_username"],
+        'password': test_config["jwt_password"],
+        'grant_type': 'password'
+    }
+
+    # Request headers
+    headers = {
+        'Content-Type': 'application/x-www-form-urlencoded'
+    }
+
+    # Make the request to get the access token
+    response = requests.post(token_url, data=payload, headers=headers)
+    tokens = response.json()
+
+    access_token = None
+    if response.status_code == 200:
+        access_token = tokens.get('access_token')
+        print(f"Access Token: {access_token}")
+    else:
+        print(f"Error fetching access token: {tokens}")
+        assert False, f"Error fetching access token: {tokens}"
+
+    results = run_query(
+        url=test_config['url'],
+        ontology=test_config['ontology'],
+        token=access_token,
+        query='SELECT 1',
+        datasource=test_config['datasource'],
+        nested='false',
+        verify_ssl=test_config['verify_ssl'],
+        enable_IPv6=test_config['enableIPv6'],
+        is_jwt=True,
+    )
+
+    assert results is not None, "Results should not be None"