feat: [sc-207225] [SAP-OData] Implement server side pagination

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## [Version 1.0.4](https://github.com/dataiku/dss-plugin-sap-odata/releases/tag/v1.0.4) - Feature - 2024-10-04
+
+- Add server side pagination
+
 ## [Version 1.0.3](https://github.com/dataiku/dss-plugin-sap-odata/releases/tag/v1.0.3) - Feature - 2023-10-20
 
 - Add filter field

Jenkinsfile

@@ -0,0 +1,47 @@
+pipeline {
+   options { disableConcurrentBuilds() }
+   agent { label 'dss-plugin-tests'}
+   environment {
+        PLUGIN_INTEGRATION_TEST_INSTANCE="/home/jenkins-agent/instance_config.json"
+    }
+   stages {
+      stage('Run Unit Tests') {
+         steps {
+            sh 'echo "Running unit tests"'
+            catchError(stageResult: 'FAILURE') {
+            sh """
+               make unit-tests
+               """
+            }
+            sh 'echo "Done with unit tests"'
+         }
+      }
+      stage('Run Integration Tests') {
+         steps {
+            sh 'echo "Running integration tests"'
+            catchError(stageResult: 'FAILURE') {
+            sh """
+               make integration-tests
+               """
+            }
+            sh 'echo "Done with integration tests"'
+         }
+      }
+   }
+   post {
+     always {
+        script {
+           allure([
+                    includeProperties: false,
+                    jdk: '',
+                    properties: [],
+                    reportBuildPolicy: 'ALWAYS',
+                    results: [[path: 'tests/allure_report']]
+            ])
+
+            def status = currentBuild.currentResult
+            sh "file_name=\$(echo ${env.JOB_NAME} | tr '/' '-').status; touch \$file_name; echo \"${env.BUILD_URL};${env.CHANGE_TITLE};${env.CHANGE_AUTHOR};${env.CHANGE_URL};${env.BRANCH_NAME};${status};\" >> $HOME/daily-statuses/\$file_name"
+        }
+     }
+   }
+}

parameter-sets/basic-auth/parameter-set.json

@@ -45,6 +45,22 @@
                 }
             ]
         },
+        {
+            "name": "sap_mode",
+            "label": "SAP mode",
+            "type": "SELECT",
+            "defaultValue" : "cds",
+            "selectChoices": [
+                {
+                    "value": "cds",
+                    "label": "CDS"
+                },
+                {
+                    "value": "odp",
+                    "label": "ODP"
+                }
+            ]
+        },
         {
             "name": "sap_client",
             "label": "Client",

parameter-sets/login/parameter-set.json

@@ -44,6 +44,22 @@
                 }
             ]
         },
+        {
+            "name": "sap_mode",
+            "label": "SAP mode",
+            "type": "SELECT",
+            "defaultValue" : "cds",
+            "selectChoices": [
+                {
+                    "value": "cds",
+                    "label": "CDS"
+                },
+                {
+                    "value": "odp",
+                    "label": "ODP"
+                }
+            ]
+        },
         {
             "name": "odata_username",
             "label": "Username",

plugin.json

@@ -1,6 +1,6 @@
 {
     "id": "sap-odata",
-    "version": "1.0.3",
+    "version": "1.0.4",
     "meta": {
         "label": "SAP OData",
         "description": "Import data from your SAP account",

python-connectors/sap_odata/connector.json

@@ -129,6 +129,7 @@
             "label": " ",
             "description": "Bulk size",
             "type": "INT",
+            "minI": 0,
             "defaultValue": 1000,
             "visibilityCondition": "model.show_advanced_parameters == true"
         }

python-connectors/sap_odata/connector.py

@@ -1,7 +1,7 @@
 from dataiku.connector import Connector
 from dataikuapi.utils import DataikuException
 from odata_client import ODataClient
-from odata_common import get_clean_row_method, get_list_title
+from odata_common import get_clean_row_method, get_list_title, RecordsLimit, get_sap_mode
 import logging
 
 
@@ -21,10 +21,11 @@ def __init__(self, config, plugin_config):
         object 'plugin_config' to the constructor
         """
         Connector.__init__(self, config, plugin_config)
-        logger.info("Starting SAP-OData v1.0.3")
+        logger.info("Starting SAP-OData v1.0.4-beta.3")
         self.odata_list_title = get_list_title(config)
         self.bulk_size = config.get("bulk_size", 1000)
         self.odata_filter_query = ""
+        self.sap_mode = get_sap_mode(config)
 
         if config.get("show_advanced_parameters", False):
             self.odata_filter_query = config.get("odata_filter_query", "")
@@ -66,30 +67,50 @@ def generate_rows(self, dataset_schema=None, dataset_partitioning=None,
 
         The dataset schema and partitioning are given for information purpose.
         """
-        skip = 0
-        bulk_size = self.bulk_size
-        if records_limit > 0:
-            bulk_size = records_limit if records_limit < bulk_size else bulk_size
+        limit = RecordsLimit(records_limit=records_limit)
+        skip = None
+        bulk_size = self.get_bulk_size(records_limit=records_limit)
         items, next_page_url = self.client.get_entity_collections(
             entity=self.odata_list_title,
             top=bulk_size,
             skip=skip,
             filter=self.odata_filter_query
         )
         while items:
+            number_of_items = len(items)
             for item in items:
                 yield self.clean_row(item)
-            skip = skip + bulk_size
-            if records_limit > 0:
-                if skip >= records_limit:
-                    break
-                if skip + bulk_size > records_limit:
-                    bulk_size = records_limit - skip
+                if limit.is_reached():
+                    logger.info("Limit is reached")
+                    return
+            if self.is_client_side_pagination():
+                if skip is None:
+                    skip = 0
+                skip = skip + number_of_items
+            else:
+                if not next_page_url:
+                    # Server side pagination with no next_page_url
+                    # -> time to quit
+                    return
             items, next_page_url = self.client.get_entity_collections(
                 entity=self.odata_list_title, top=bulk_size, skip=skip,
-                page_url=next_page_url, filter=self.odata_filter_query
+                page_url=next_page_url, filter=self.odata_filter_query, can_raise=False
             )
 
+    def get_bulk_size(self, records_limit=None):
+        if self.is_client_side_pagination():
+            if self.bulk_size == 0:
+                return None
+            bulk_size = self.bulk_size
+            if records_limit > 0:
+                bulk_size = records_limit if records_limit < bulk_size else bulk_size
+        else:
+            bulk_size = None
+        return bulk_size
+
+    def is_client_side_pagination(self):
+        return self.sap_mode == "cds"
+
     def get_schema_set(self, set_name):
         for one_set in self.client.schema.entity_sets:
             if one_set.name == set_name:

python-lib/odata_client.py

@@ -84,18 +84,25 @@ def get_session(self, config, odata_version):
             )
         return session
 
-    def get_entity_collections(self, entity="", top=None, skip=None, page_url=None, filter=None):
+    def get_entity_collections(self, entity="", top=None, skip=None, page_url=None, filter=None, can_raise=True):
+        if entity is None:
+            entity = ""
         if self.odata_list_title is None or self.odata_list_title == "":
             top = None  # SAP will complain if $top is present in a request to list entities
         query_options = self.get_base_query_options(top=top, skip=skip, filter=filter)
         url = page_url if page_url else self.odata_instance + '/' + entity.strip("/") + self.get_query_string(query_options)
         data = None
         while self._should_retry(data):
+            logger.info("requests get url {}".format(url))
             response = self.get(url)
-            self.assert_response(response)
+            if self.assert_response_ok(response, can_raise=can_raise):
+                data = response.json()
+            else:
+                return {}, None
             data = response.json()
         next_page_url = data.get(ODataConstants.NEXT_LINK, None)
         item = data.get(ODataConstants.DATA_CONTAINER_V4, data.get(ODataConstants.DATA_CONTAINER_V2, {}))
+        next_page_url = item.get(ODataConstants.NEXT_LINK_SAP, next_page_url)
         return self.format(item), next_page_url
 
     def _should_retry(self, data):
@@ -184,11 +191,21 @@ def get_query_string(self, query_options):
         else:
             return ""
 
-    def assert_response(self, response):
+    def assert_response_ok(self, response, can_raise=True):
         status_code = response.status_code
+        return_code = True
         if status_code == 404:
-            raise DataikuException("This entity does not exist")
+            return_code = False
+            logger.error("Error 404, response={}".format(response.content))
+            if can_raise:
+                raise DataikuException("This entity does not exist")
         if status_code == 403:
             raise DataikuException("{}".format(response))
         if status_code == 401:
             raise DataikuException("Forbidden access")
+        if status_code == 400:
+            return_code = False
+            logger.error("Error 400, response={}".format(response.content))
+            if can_raise:
+                raise DataikuException("Error 400: {}".format(response))
+        return return_code

python-lib/odata_common.py

@@ -83,6 +83,13 @@ def get_list_title(config):
     return odata_list_title
 
 
+def get_sap_mode(config):
+    auth_type = config.get("auth_type", "login")
+    login_config = config.get(ODataConstants.LOGIN, {}) if auth_type == "login" else config.get("sap-odata_user-account", {})
+    sap_mode = login_config.get("sap_mode", "cds")
+    return sap_mode
+
+
 class DSSSelectorChoices(object):
     def __init__(self):
         self.choices = []
@@ -118,3 +125,16 @@ def text_message(self, text_message):
 
     def to_dss(self):
         return self._build_select_choices(self.choices)
+
+
+class RecordsLimit():
+    def __init__(self, records_limit=-1):
+        self.has_no_limit = (records_limit == -1)
+        self.records_limit = records_limit
+        self.counter = 0
+
+    def is_reached(self):
+        if self.has_no_limit:
+            return False
+        self.counter += 1
+        return self.counter > self.records_limit

python-lib/odata_constants.py

@@ -9,6 +9,7 @@ class ODataConstants(object):
     LIST_TITLE = "odata_list_title"
     LOGIN = "sap-odata_login"
     NEXT_LINK = "@odata.nextLink"
+    NEXT_LINK_SAP = "__next"
     OAUTH = "odata_oauth"
     ODATA_V2 = "v2"
     ODATA_V3 = "v3"

tests/python/integration/pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+usefixtures = plugin dss_target

tests/python/integration/requirements.txt

@@ -0,0 +1,3 @@
+pytest==6.2.1
+dataiku-api-client
+git+git://github.com/dataiku/dataiku-plugin-tests-utils.git@master#egg=dataiku-plugin-tests-utils

tests/python/integration/test_scenario.py

@@ -0,0 +1,12 @@
+from dku_plugin_test_utils import dss_scenario
+
+
+TEST_PROJECT_KEY = "PLUGINTESTSAPODATA"
+
+
+def test_run_sap_odata_read(user_dss_clients):
+    dss_scenario.run(user_dss_clients, project_key=TEST_PROJECT_KEY, scenario_id="SAP")
+
+
+def test_run_sap_odata_v4(user_dss_clients):
+    dss_scenario.run(user_dss_clients, project_key=TEST_PROJECT_KEY, scenario_id="SAP_V4")