Add a switch for selecting pagination traversal in edx_api_utils

openedx/core/lib/edx_api_utils.py

@@ -14,8 +14,8 @@ from openedx.core.lib.token_utils import JwtBuilder
 log = logging.getLogger(__name__)
 
 
-def get_edx_api_data(api_config, user, resource,
-                     api=None, resource_id=None, querystring=None, cache_key=None, many=True):
+def get_edx_api_data(api_config, user, resource, api=None, resource_id=None,
+                     querystring=None, cache_key=None, many=True, traverse_pagination=True):
     """GET data from an edX REST API.
 
     DRY utility for handling caching and pagination.
@@ -33,6 +33,7 @@ def get_edx_api_data(api_config, user, resource,
             (neither inspected nor updated).
         many (bool): Whether the resource requested is a collection of objects, or a single object.
             If false, an empty dict will be returned in cases of failure rather than the default empty list.
+        traverse_pagination (bool): Whether to traverse pagination or return paginated response..
 
     Returns:
         Data returned by the API. When hitting a list endpoint, extracts "results" (list of dict)
@@ -79,8 +80,10 @@ def get_edx_api_data(api_config, user, resource,
 
         if resource_id is not None:
             results = response
-        else:
+        elif traverse_pagination:
             results = _traverse_pagination(response, endpoint, querystring, no_data)
+        else:
+            results = response
     except:  # pylint: disable=bare-except
         log.exception('Failed to retrieve data from the %s API.', api_config.API_NAME)
         return no_data

openedx/core/lib/tests/test_edx_api_utils.py

@@ -101,6 +101,36 @@ class TestGetEdxApiData(CatalogIntegrationMixin, CredentialsApiConfigMixin, Cach
 
         self._assert_num_requests(len(expected_collection))
 
+    def test_get_paginated_data_do_not_traverse_pagination(self):
+        """
+        Verify that pagination is not traversed if traverse_pagination=False is passed as argument.
+        """
+        catalog_integration = self.create_catalog_integration()
+        api = create_catalog_api_client(self.user, catalog_integration)
+
+        url = CatalogIntegration.current().internal_api_url.strip('/') + '/programs/?page={}'
+        responses = [
+            {
+                'next': url.format(2),
+                'results': ['some'],
+            },
+            {
+                'next': url.format(None),
+                'results': ['test'],
+            },
+        ]
+        expected_response = responses[0]
+
+        self._mock_catalog_api(
+            [httpretty.Response(body=json.dumps(body), content_type='application/json') for body in responses]
+        )
+
+        actual_collection = get_edx_api_data(
+            catalog_integration, self.user, 'programs', api=api, traverse_pagination=False,
+        )
+        self.assertEqual(actual_collection, expected_response)
+        self._assert_num_requests(1)
+
     def test_get_specific_resource(self):
         """Verify that a specific resource can be retrieved."""
         catalog_integration = self.create_catalog_integration()