Merge pull request #24 from dataiku/bug/sc-98256-offset-pagination-on-lists

[sc-98256] Handling offset pagination on APIs returning an array

Author: alexbourret

CHANGELOG.md

@@ -1,6 +1,13 @@
 # Changelog
 
 
+## [Version 1.1.0](https://github.com/dataiku/dss-plugin-api-connect/releases/tag/v1.1.0) - Feature and bugfix release - 2022-09-15
+
+- Handling Offset pagination on APIs returning an array
+- Fix throttling calculation
+- Allow dots in `Key to next request URL`
+- Add NTLM authentication
+
 ## [Version 1.0.6](https://github.com/dataiku/dss-plugin-api-connect/releases/tag/v1.0.6) - Feature and bugfix release - 2022-05-19
 
 - Add "Follow authorization header" option

code-env/python/spec/requirements.txt

@@ -0,0 +1,2 @@
+jsonpath-ng==1.5.3
+requests_ntlm==1.1.0

parameter-sets/credential/parameter-set.json

@@ -38,6 +38,10 @@
                 {
                     "value": "bearer_token",
                     "label": "Bearer token"
+                },
+                {
+                    "value": "ntlm",
+                    "label": "NTLM"
                 }
             ]
         },
@@ -46,14 +50,14 @@
             "label": "User name",
             "description": "Can be reused as {{username}}",
             "type": "STRING",
-            "visibilityCondition": "model.login_type == 'basic_login'"
+            "visibilityCondition": "['basic_login', 'ntlm'].includes(model.login_type)"
         },
         {
             "name": "password",
             "label": "Password",
             "description": "Can be reused as {{password}}",
             "type": "PASSWORD",
-            "visibilityCondition": "model.login_type == 'basic_login'"
+            "visibilityCondition": "['basic_login', 'ntlm'].includes(model.login_type)"
         },
         {
             "name": "token",

plugin.json

@@ -1,6 +1,6 @@
 {
     "id": "api-connect",
-    "version": "1.0.6",
+    "version": "1.1.0",
     "meta": {
         "label": "API Connect",
         "description": "Retrieve data from any REST API",

python-lib/dku_utils.py

@@ -1,5 +1,6 @@
 import json
 import copy
+from jsonpath_ng.ext import parse
 
 
 def get_dku_key_values(endpoint_query_string):
@@ -46,7 +47,7 @@ def parse_keys_for_json(items):
 def get_value_from_path(dictionary, path, default=None, can_raise=True):
     ret = copy.deepcopy(dictionary)
     for key in path:
-        if key in ret and isinstance(ret, dict):
+        if isinstance(ret, dict) and (key in ret):
             ret = ret.get(key)
         else:
             error_message = "The extraction path {} was not found in the incoming data".format(path)
@@ -84,3 +85,12 @@ def format_template(template, **kwargs):
 def is_string(data):
     data_type = type(data).__name__
     return data_type in ["str", "unicode"]
+
+
+def extract_key_using_json_path(json_dictionary, json_path):
+    matches = parse(json_path).find(json_dictionary)
+    if matches:
+        res = matches[0].value
+        return res
+    else:
+        return None

python-lib/pagination.py

@@ -1,5 +1,5 @@
 from safe_logger import SafeLogger
-from dku_utils import get_value_from_path
+from dku_utils import get_value_from_path, extract_key_using_json_path
 
 
 logger = SafeLogger("api-connect plugin Pagination")
@@ -24,14 +24,15 @@ def __init__(self, config=None, skip_key=None, limit_key=None, total_key=None, n
         self.is_paging_started = None
         self.next_page_number = None
         self.params_must_be_blanked = False
+        self.data_is_list = None
 
     def configure_paging(self, config=None, skip_key=None, limit_key=None, total_key=None, next_page_key=None, url=None, pagination_type="na"):
         config = {} if config is None else config
         self.limit_key = config.get("limit_key", limit_key)
         self.pagination_type = config.get("pagination_type", pagination_type)
         if self.pagination_type == "next_page":
             self.next_page_key = config.get("next_page_key", next_page_key)
-            self.next_page_key = None if self.next_page_key == [''] else self.next_page_key
+            self.next_page_key = None if self.next_page_key == '' else self.next_page_key
         elif self.pagination_type in ["offset", "page"]:
             self.skip_key = config.get("skip_key", skip_key)
 
@@ -62,7 +63,9 @@ def update_next_page(self, data, response_links=None):
         if next_page_url:
             self.next_page_url = next_page_url
             self.params_must_be_blanked = True
+        self.data_is_list = False
         if isinstance(data, list):
+            self.data_is_list = True
             batch_size = len(data)
             self.records_to_skip = self.records_to_skip + batch_size
             if batch_size == 0:
@@ -77,8 +80,8 @@ def update_next_page(self, data, response_links=None):
                 self.is_last_batch_empty = True
         else:
             batch_size = 1
-        if self.next_page_key and (len(self.next_page_key) > 0):
-            self.next_page_url = self.get_from_path(data, self.next_page_key)
+        if self.next_page_key:
+            self.next_page_url = extract_key_using_json_path(data, self.next_page_key)
         if self.skip_key:
             self.skip = data.get(self.skip_key)
         if self.limit_key:
@@ -89,17 +92,6 @@ def update_next_page(self, data, response_links=None):
         if self.total:
             self.remaining_records = self.total - self.records_to_skip
 
-    def get_from_path(self, dictionary, path):
-        if isinstance(path, list):
-            edge = dictionary
-            for key in path:
-                edge = edge.get(key)
-                if edge is None:
-                    return None
-            return edge
-        else:
-            return dictionary.get(path)
-
     def has_next_page(self):
         if self.is_last_batch_empty:
             logger.info("has_next_page:last was batch empty -> False")
@@ -120,7 +112,11 @@ def has_next_page(self):
                 #  There is a counting key and we already know the last batch was not empty
                 return True
             else:
-                #  No way to know if the last batch was empty so we stop here
+                if self.data_is_list:
+                    # for lists is_last_batch_empty is set correctly and handled by the code above
+                    return True
+                # Without a counting_key we have no mean to know if the last batch was empty.
+                # To avoid infinite loop we stop pagination here
                 return False
         return False
 

python-lib/rest_api_client.py

@@ -60,7 +60,7 @@ def __init__(self, credential, endpoint, custom_key_values={}):
 
         self.requests_kwargs.update({"params": self.params})
         self.pagination = Pagination()
-        next_page_url_key = endpoint.get("next_page_url_key", "").split('.')
+        next_page_url_key = endpoint.get("next_page_url_key", "")
         top_key = endpoint.get("top_key")
         skip_key = endpoint.get("skip_key")
         pagination_type = endpoint.get("pagination_type", "na")
@@ -87,14 +87,21 @@ def __init__(self, credential, endpoint, custom_key_values={}):
             key_value_body = endpoint.get("key_value_body", {})
             self.requests_kwargs.update({"json": get_dku_key_values(key_value_body)})
         self.metadata = {}
+        self.call_number = 0
 
     def set_login(self, credential):
         login_type = credential.get("login_type", "no_auth")
         if login_type == "basic_login":
-            self.username = credential.get("username", "")
-            self.password = credential.get("password", "")
-            self.auth = (self.username, self.password)
-            self.requests_kwargs.update({"auth": self.auth})
+            username = credential.get("username", "")
+            password = credential.get("password", "")
+            auth = (username, password)
+            self.requests_kwargs.update({"auth": auth})
+        if login_type == "ntlm":
+            from requests_ntlm import HttpNtlmAuth
+            username = credential.get("username", "")
+            password = credential.get("password", "")
+            auth = HttpNtlmAuth(username, password)
+            self.requests_kwargs.update({"auth": auth})
         if login_type == "bearer_token":
             token = credential.get("token", "")
             bearer_template = credential.get("bearer_template", "Bearer {{token}}")
@@ -119,8 +126,9 @@ def request(self, method, url, can_raise_exeption=True, **kwargs):
         kwargs = template_dict(kwargs, **self.presets_variables)
         if self.loop_detector.is_stuck_in_loop(url, kwargs.get("params", {}), kwargs.get("headers", {})):
             raise RestAPIClientError("The api-connect plugin is stuck in a loop. Please check the pagination parameters.")
+        request_start_time = time.time()
+        self.time_last_request = request_start_time
         try:
-            request_start_time = time.time()
             response = self.request_with_redirect_retry(method, url, **kwargs)
             request_finish_time = time.time()
         except Exception as err:
@@ -131,7 +139,6 @@ def request(self, method, url, can_raise_exeption=True, **kwargs):
             else:
                 return {"error": error_message}
         self.set_metadata("request_duration", request_finish_time - request_start_time)
-        self.time_last_request = time.time()
         self.set_metadata("status_code", response.status_code)
         self.set_metadata("response_headers", "{}".format(response.headers))
         if response.status_code >= 400:
@@ -167,6 +174,8 @@ def paginated_api_call(self, can_raise_exeption=True):
             params = self.requests_kwargs.get("params")
             params.update(pagination_params)
             self.requests_kwargs.update({"params": params})
+        self.call_number = self.call_number + 1
+        logger.info("API call number #{}".format(self.call_number))
         return self.request(self.http_method, self.pagination.get_next_page_url(), can_raise_exeption, **self.requests_kwargs)
 
     def empty_json_response(self):

python-lib/rest_api_recipe_session.py

@@ -46,7 +46,8 @@ def process_dataframe(self, input_parameters_dataframe, is_raw_output):
                 self.initial_parameter_columns.update({parameter_name: input_parameters_row.get(column_name)})
             updated_endpoint_parameters = copy.deepcopy(self.endpoint_parameters)
             updated_endpoint_parameters.update(self.initial_parameter_columns)
-            logger.info("Creating client with credential={}, updated_endpoint={}, custom_key_values={}".format(
+            logger.info("Processing row #{}, creating client with credential={}, updated_endpoint={}, custom_key_values={}".format(
+                index + 1,
                 logger.filter_secrets(self.credential_parameters),
                 updated_endpoint_parameters,
                 self.custom_key_values

tests/python/integration/test_scenario.py

@@ -33,3 +33,6 @@ def test_run_api_connect_redirection(user_dss_clients):
 
 def test_run_api_connect_check_sc_84465(user_dss_clients):
     dss_scenario.run(user_dss_clients, project_key=TEST_PROJECT_KEY, scenario_id="CHECKSC84465")
+
+def test_run_api_connect_ntlm_authentication(user_dss_clients):
+    dss_scenario.run(user_dss_clients, project_key=TEST_PROJECT_KEY, scenario_id="NTLMAUTHENTICATION")