Add support for auto-paginated Admin API calls

* Add support for auto-paginated API calls

* Simplify auto pagination usage

* Remove unused import

* Add tests

* Update help string

* Rename known_field to paginate_field

* Align code with SearchApi

Co-authored-by: Brian Luk <[email protected]>
Co-authored-by: Constantine Nathanson <[email protected]>

Author: 3 people

cloudinary_cli/core/admin.py

@@ -19,11 +19,20 @@
 @option("-o", "--optional_parameter", multiple=True, nargs=2, help="Pass optional parameters as raw strings.")
 @option("-O", "--optional_parameter_parsed", multiple=True, nargs=2,
         help="Pass optional parameters as interpreted strings.")
+@option("-A", "--auto_paginate", is_flag=True, help="Will auto paginate Admin API calls.", default=False)
+@option("-ff", "--filter_fields", multiple=True, help="Filter fields to return when using auto pagination.")
+@option("-F", "--force", is_flag=True, help="Skip confirmation when running --auto-paginate.")
 @option("-ls", "--ls", is_flag=True, help="List all available methods in the Admin API.")
 @option("--save", nargs=1, help="Save output to a file.")
 @option("-d", "--doc", is_flag=True, help="Open the Admin API reference in a browser.")
-def admin(params, optional_parameter, optional_parameter_parsed, ls, save, doc):
-    return handle_api_command(params, optional_parameter, optional_parameter_parsed, ls, save, doc,
+def admin(params, optional_parameter, optional_parameter_parsed,
+          auto_paginate, force, filter_fields,
+          ls, save, doc):
+    return handle_api_command(params, optional_parameter, optional_parameter_parsed,
+                              ls, save, doc,
                               doc_url="https://cloudinary.com/documentation/admin_api",
                               api_instance=api,
-                              api_name="admin")
+                              api_name="admin",
+                              auto_paginate=auto_paginate,
+                              force=force,
+                              filter_fields=filter_fields)

cloudinary_cli/core/search.py

@@ -5,7 +5,8 @@
 
 from cloudinary_cli.defaults import logger
 from cloudinary_cli.utils.json_utils import write_json_to_file, print_json
-from cloudinary_cli.utils.utils import write_json_list_to_csv, confirm_action
+from cloudinary_cli.utils.utils import write_json_list_to_csv, confirm_action, whitelist_keys, \
+    normalize_list_params
 
 DEFAULT_MAX_RESULTS = 500
 
@@ -38,12 +39,7 @@ def search(query, with_field, sort_by, aggregate, max_results, next_cursor,
 
     fields_to_keep = []
     if filter_fields:
-        for f in list(filter_fields):
-            if "," in f:
-                fields_to_keep += f.split(",")
-            else:
-                fields_to_keep.append(f)
-        fields_to_keep = tuple(fields_to_keep) + with_field
+        fields_to_keep = tuple(normalize_list_params(filter_fields)) + with_field
 
     expression = cloudinary.search.Search().expression(" ".join(query))
 
@@ -81,15 +77,13 @@ def execute_single_request(expression, fields_to_keep):
     res = expression.execute()
 
     if fields_to_keep:
-        res['resources'] = list(
-            map(lambda x: {k: x[k] if k in x.keys() else None for k in fields_to_keep}, res['resources'])
-        )
+        res['resources'] = whitelist_keys(res['resources'], fields_to_keep)
 
     return res
 
 
 def handle_auto_pagination(res, expression, force, fields_to_keep):
-    if 'next_cursor' not in res.keys():
+    if 'next_cursor' not in res:
         return res
 
     if not force:
@@ -99,6 +93,7 @@ def handle_auto_pagination(res, expression, force, fields_to_keep):
                 f"Running this query will use {res['total_count'] // DEFAULT_MAX_RESULTS + 1} Admin API calls. "
                 f"Continue? (y/N)"):
             logger.info("Stopping. Please run again without -A.")
+
             return res
         else:
             logger.info("Continuing. You may use the -F flag to force auto_pagination.")
@@ -112,6 +107,6 @@ def handle_auto_pagination(res, expression, force, fields_to_keep):
         all_results['resources'] += res['resources']
         all_results['time'] += res['time']
 
-    all_results.pop('next_cursor')  # it is empty by now
+    all_results.pop('next_cursor', None)  # it is empty by now
 
     return all_results

cloudinary_cli/utils/api_utils.py

@@ -9,7 +9,12 @@
 
 from cloudinary_cli.defaults import logger
 from cloudinary_cli.utils.json_utils import print_json, write_json_to_file
-from cloudinary_cli.utils.utils import print_help, parse_args_kwargs, parse_option_value, log_exception
+from cloudinary_cli.utils.utils import print_help, log_exception, confirm_action, \
+    get_command_params, merge_responses, normalize_list_params
+
+PAGINATION_MAX_RESULTS = 500
+
+_cursor_fields = {"resource": "derived_next_cursor"}
 
 
 def query_cld_folder(folder):
@@ -93,28 +98,22 @@ def asset_source(asset_details):
     return base_name + '.' + asset_details['format']
 
 
+def call_api(func, args, kwargs):
+    return func(*args, **kwargs)
+
+
 def handle_command(
         params,
         optional_parameter,
         optional_parameter_parsed,
         module,
         module_name):
-    try:
-        func = module.__dict__[params[0]]
-    except KeyError:
-        raise Exception(f"Method {params[0]} does not exist in {module_name.capitalize()}.")
-
-    if not callable(func):
-        raise Exception(f"{params[0]} is not callable.")
-
-    args, kwargs = parse_args_kwargs(func, params[1:]) if len(params) > 1 else ([], {})
-    kwargs = {
-        **kwargs,
-        **{k: v for k, v in optional_parameter},
-        **{k: parse_option_value(v) for k, v in optional_parameter_parsed},
-    }
-
-    return func(*args, **kwargs)
+    func, args, kwargs = get_command_params(params,
+                                            optional_parameter,
+                                            optional_parameter_parsed,
+                                            module,
+                                            module_name)
+    return call_api(func, args, kwargs)
 
 
 def handle_api_command(
@@ -126,7 +125,10 @@ def handle_api_command(
         doc,
         doc_url,
         api_instance,
-        api_name):
+        api_name,
+        auto_paginate=False,
+        force=False,
+        filter_fields=None):
     """
     Used by Admin and Upload API commands
     """
@@ -136,14 +138,56 @@ def handle_api_command(
     if ls or len(params) < 1:
         return print_help(api_instance)
 
-    res = handle_command(
+    func, args, kwargs = get_command_params(
         params,
         optional_parameter,
         optional_parameter_parsed,
         api_instance,
         api_name)
 
+    res = call_api(func, args, kwargs)
+
+    if auto_paginate:
+        res = handle_auto_pagination(res, func, args, kwargs, force, filter_fields)
+
     print_json(res)
 
     if save:
         write_json_to_file(res, save)
+
+
+def handle_auto_pagination(res, func, args, kwargs, force, filter_fields):
+    cursor_field = _cursor_fields.get(func.__name__, "next_cursor")
+
+    if cursor_field not in res:
+        return res
+
+    if not force:
+        if not confirm_action(
+                "Using auto pagination will use multiple API calls.\n" +
+                f"You currently have {res.rate_limit_remaining} Admin API calls remaining. Continue? (y/N)"):
+            logger.info("Stopping. Please run again without -A.")
+
+            return res
+        else:
+            logger.info("Continuing. You may use the -F flag to force auto_pagination.")
+
+    fields_to_keep = []
+    if filter_fields:
+        fields_to_keep = normalize_list_params(filter_fields)
+
+    kwargs['max_results'] = PAGINATION_MAX_RESULTS
+
+    all_results = res
+    # We have many different APIs that have different fields that we paginate.
+    # The field is unknown before we perform the second call and then compare results and find the field.
+    pagination_field = None
+    while res.get(cursor_field, None):
+        kwargs[cursor_field] = res.get(cursor_field, None)
+        res = call_api(func, args, kwargs)
+        all_results, pagination_field = merge_responses(all_results, res, fields_to_keep=fields_to_keep,
+                                                        pagination_field=pagination_field)
+
+    all_results.pop(cursor_field, None)
+
+    return all_results

cloudinary_cli/utils/utils.py

@@ -154,3 +154,93 @@ def get_user_action(message, options):
     """
     r = input(message).lower()
     return options.get(r, options.get("default"))
+
+
+def get_command_params(
+        params,
+        optional_parameter,
+        optional_parameter_parsed,
+        module,
+        module_name):
+    method = params[0]
+    if method not in module.__dict__:
+        raise Exception(f"Method {params[0]} does not exist in {module_name.capitalize()}.")
+
+    func = module.__dict__.get(method)
+
+    if not callable(func):
+        raise Exception(f"{params[0]} is not callable.")
+
+    args, kwargs = parse_args_kwargs(func, params[1:]) if len(params) > 1 else ([], {})
+
+    kwargs = {
+        **kwargs,
+        **{k: v for k, v in optional_parameter},
+        **{k: parse_option_value(v) for k, v in optional_parameter_parsed},
+    }
+
+    return func, args, kwargs
+
+
+def whitelist_keys(data, keys):
+    """
+    Iterates over a list of dictionaries and keeps only the keys that were specified.
+
+    :param data: A list of dictionaries.
+    :type data: list
+    :param keys: a list of keys to keep in each dictionary.
+    :type keys: list
+
+    :return: The whitelisted list.
+    :rtype list
+    """
+    # no whitelist when fields are not provided or on a list of non-dictionary items.
+    if not keys or any(not isinstance(i, dict) for i in data):
+        return data
+
+    return list(
+        map(lambda x: {
+            k: x[k]
+            for k in keys if k in x},
+            data)
+    )
+
+
+def merge_responses(all_res, paginated_res, fields_to_keep=None, pagination_field=None):
+    if not pagination_field:
+        for key in all_res:
+            if all_res[key] != paginated_res.get(key, 0) and type(all_res[key]) == list:
+                pagination_field = key
+
+        if not pagination_field:  # should not happen
+            raise Exception("Failed to detect pagination_field")
+
+        # whitelist fields of the initial response
+        all_res[pagination_field] = whitelist_keys(all_res[pagination_field], fields_to_keep)
+
+    all_res[pagination_field] += whitelist_keys(paginated_res[pagination_field], fields_to_keep)
+
+    return all_res, pagination_field
+
+
+def normalize_list_params(params):
+    """
+    Normalizes parameters that could be provided as strings separated by ','.
+
+    >>> normalize_list_params(["f1,f2", "f3"])
+    ["f1", "f2", "f3"]
+
+    :param params: Params to normalize.
+    :type params: list
+
+    :return: A list of normalized params.
+    :rtype list
+    """
+    normalized_params = []
+    for f in list(params):
+        if "," in f:
+            normalized_params += f.split(",")
+        else:
+            normalized_params.append(f)
+
+    return normalized_params

test/test_utils.py

@@ -1,6 +1,6 @@
 import unittest
 
-from cloudinary_cli.utils.utils import parse_option_value
+from cloudinary_cli.utils.utils import parse_option_value, whitelist_keys, merge_responses, normalize_list_params
 
 
 class UtilsTest(unittest.TestCase):
@@ -18,3 +18,46 @@ def test_parse_option_value(self):
     def test_parse_option_value_converts_int_to_str(self):
         """ should convert a parsed int to a str """
         self.assertEqual("1", parse_option_value(1))
+
+    def test_whitelist_keys(self):
+        """ should whitelist keys correctly """
+        self.assertEqual([{"k1": "v1"}], whitelist_keys([{"k1": "v1", "k2": "v2"}], ["k1"]))
+        self.assertEqual([{"k1": "v1", "k2": "v2"}], whitelist_keys([{"k1": "v1", "k2": "v2"}], []))
+        self.assertEqual([{"k1": "v1"}], whitelist_keys([{"k1": "v1", "k2": "v2"}], ["k1", "k3"]))
+
+    def test_merge_responses(self):
+        """ should merge responses based with or without additional kwargs """
+
+        merged_1 = ({"a": "b", "c": [{"1": "2"}, {"1": "3"}, {"1": "4"}, {"1": "5"}]}, "c")
+        merged_1_2 = (
+            {"a": "b", "c": [{"1": "2", "2": "2"}, {"1": "3", "2": "2"}, {"1": "4", "2": "2"}, {"1": "5", "2": "2"}]},
+            "c"
+        )
+        self.assertEqual(
+            merged_1,
+            merge_responses(
+                {"a": "b", "c": [{"1": "2", "2": "2"}, {"1": "3", "2": "2"}]},
+                {"a": "b", "c": [{"1": "4", "2": "2"}, {"1": "5", "2": "2"}]},
+                fields_to_keep=["1"]))
+        self.assertEqual(
+            merged_1,
+            merge_responses(
+                {"a": "b", "c": [{"1": "2"}, {"1": "3"}]},
+                {"a": "b", "c": [{"1": "4", "2": "2"}, {"1": "5", "2": "2"}]},
+                fields_to_keep=["1"],
+                pagination_field="c"))
+        self.assertEqual(
+            merged_1_2,
+            merge_responses(
+                {"a": "b", "c": [{"1": "2", "2": "2"}, {"1": "3", "2": "2"}]},
+                {"a": "b", "c": [{"1": "4", "2": "2"}, {"1": "5", "2": "2"}]},
+                pagination_field="c"))
+        self.assertEqual(
+            merged_1_2,
+            merge_responses(
+                {"a": "b", "c": [{"1": "2", "2": "2"}, {"1": "3", "2": "2"}]},
+                {"a": "b", "c": [{"1": "4", "2": "2"}, {"1": "5", "2": "2"}]}))
+
+    def test_normalize_list_params(self):
+        """ should normalize a list of parameters """
+        self.assertEqual(["f1", "f2", "f3"], normalize_list_params(["f1,f2", "f3"]))