Allow passing positional parameters as optional

Author: const-cloudinary

cloudinary_cli/modules/upload_dir.py

@@ -6,7 +6,7 @@
 
 from cloudinary_cli.utils.api_utils import upload_file
 from cloudinary_cli.utils.file_utils import get_destination_folder, is_hidden_path
-from cloudinary_cli.utils.utils import parse_option_value, logger, run_tasks_concurrently
+from cloudinary_cli.utils.utils import parse_option_value, logger, run_tasks_concurrently, group_params
 
 
 @command("upload_dir", help="""Upload a folder of assets, maintaining the folder structure.""")
@@ -62,8 +62,7 @@ def upload_dir(directory, glob_pattern, include_hidden, optional_parameter, opti
 
     options = {
         **defaults,
-        **{k: v for k, v in optional_parameter},
-        **{k: parse_option_value(v) for k, v in optional_parameter_parsed},
+        **group_params(optional_parameter, ((k, parse_option_value(v)) for k, v in optional_parameter_parsed)),
     }
 
     uploads = []

cloudinary_cli/utils/utils.py

@@ -95,6 +95,9 @@ def load_template(language, template_name):
 
 
 def parse_option_value(value):
+    if isinstance(value, list):
+        return list(map(parse_option_value, value))
+
     if value == "True" or value == "true":
         return True
     elif value == "False" or value == "false":
@@ -109,20 +112,27 @@ def parse_option_value(value):
     return value
 
 
-def parse_args_kwargs(func, params):
+def parse_args_kwargs(func, params=None, kwargs=None):
+    if params is None:
+        params = []
+    if kwargs is None:
+        kwargs = {}
+
     spec = getfullargspec(func)
-    n_args = len(spec.args) if spec.args else 0
-    n_defaults = len(spec.defaults) if spec.defaults else 0
 
-    n_req = n_args - n_defaults
-    if len(params) < n_req:
+    num_args = len(spec.args) if spec.args else 0
+    num_defaults = len(spec.defaults) if spec.defaults else 0
+
+    num_req = num_args - num_defaults
+    num_provided_args = len(params)
+    num_overall_provided = num_provided_args + len([p for p in kwargs.keys() if p in spec.args[num_provided_args:]])
+    if num_overall_provided < num_req:
         func_sig = signature(func)
-        raise Exception(f"Function '{func.__name__}{func_sig}' requires {n_req} positional arguments")
+        raise Exception(f"Function '{func.__name__}{func_sig}' requires {num_req} positional arguments")
     # consume required args
-    args = [parse_option_value(p) for p in params[:n_req]]
-    kwargs = {}
+    args = [parse_option_value(p) for p in params[:num_req]]
 
-    for p in params[n_req:]:
+    for p in params[num_req:]:
         if '=' not in p:
             # named/positional with default value args passed as positional
             args.append(parse_option_value(p))
@@ -207,17 +217,31 @@ def get_command_params(
     if not callable(func):
         raise Exception(f"{params[0]} is not callable.")
 
-    args, kwargs = parse_args_kwargs(func, params[1:])
+    kwargs = group_params(optional_parameter, ((k, parse_option_value(v)) for k, v in optional_parameter_parsed))
 
-    kwargs = {
-        **kwargs,
-        **{k: v for k, v in optional_parameter},
-        **{k: parse_option_value(v) for k, v in optional_parameter_parsed},
-    }
+    args, kwargs = parse_args_kwargs(func, params[1:], kwargs)
 
     return func, args, kwargs
 
 
+def group_params(*params):
+    """
+    Groups parameters (which are passed as list of tuples) by keys. Duplicate keys' values are combined into lists.
+
+    :param params: the list of parameters to group
+    :return: dict
+    """
+    res = {}
+    for param in params:
+        for k, v in param:
+            if k in res:
+                res[k] = (res[k] if isinstance(res[k], list) else [res[k]]) + [v]
+                continue
+            res[k] = v
+
+    return res
+
+
 def print_help_and_exit():
     """
     Prints help for the current command and exits.

test/test_utils.py

@@ -1,7 +1,7 @@
 import unittest
 
 from cloudinary_cli.utils.utils import parse_option_value, parse_args_kwargs, whitelist_keys, merge_responses, \
-    normalize_list_params, chunker
+    normalize_list_params, chunker, group_params
 
 
 class UtilsTest(unittest.TestCase):
@@ -13,8 +13,14 @@ def test_parse_option_value(self):
         self.assertListEqual(["test", "123"], parse_option_value('["test","123"]'))
         self.assertDictEqual({"foo": "bar"}, parse_option_value('{"foo":"bar"}'))
         self.assertDictEqual({"an": "object", "or": "dict"}, parse_option_value('{"an":"object","or":"dict"}'))
-        self.assertListEqual(["this", "will", "be", "read", "as",
-                              "a", "list"], parse_option_value('["this","will","be","read","as","a","list"]'))
+        self.assertListEqual(
+            ["this", "will", "be", "read", "as","a", "list"],
+            parse_option_value('["this","will","be","read","as","a","list"]')
+        )
+        self.assertListEqual(
+            [True, False, 123, '0', ['test', '123']],
+            parse_option_value(["True", "false", "123", "0", '["test","123"]'])
+        )
 
     def test_parse_option_value_converts_int_to_str(self):
         """ should convert a parsed 0 to a str """
@@ -47,6 +53,25 @@ def test_parse_args_kwargs(self):
         self.assertListEqual(["a1", "a2"], args)
         self.assertEqual(0, len(kwargs))
 
+        # should allow passing positional as optional
+        args, kwargs = parse_args_kwargs(_only_args_test_func, ["a1"], {"arg2": "a2"})
+        self.assertListEqual(["a1"], args)
+        self.assertDictEqual({"arg2": "a2"}, kwargs)
+
+        args, kwargs = parse_args_kwargs(_only_args_test_func, [], {"arg1": "a1", "arg2": "a2"})
+        self.assertEqual(0, len(args))
+        self.assertDictEqual({"arg1": "a1", "arg2": "a2"}, kwargs)
+
+    def test_group_params(self):
+        self.assertDictEqual({}, group_params([]))
+        self.assertDictEqual({"k1": "v1", "k2": "v2"}, group_params([("k1", "v1"), ("k2", "v2")]))
+        self.assertDictEqual({"k1": ["v1", "v2"]}, group_params([("k1", "v1"), ("k1", "v2")]))
+        self.assertDictEqual({"k1": "v1", "k2": ["v2", "v3"]}, group_params([("k1", "v1"), ("k2", "v2"), ("k2", "v3")]))
+        self.assertDictEqual(
+            {"k1": ["v1", "v2", "v3"]},
+            group_params([("k1", "v1")], [("k1", "v2")], [("k1", "v3")])
+        )
+
     def test_whitelist_keys(self):
         """ should whitelist keys correctly """
         self.assertEqual([{"k1": "v1"}], whitelist_keys([{"k1": "v1", "k2": "v2"}], ["k1"]))