Add support for utility methods

Author: const-cloudinary

cloudinary_cli/core/__init__.py

@@ -4,7 +4,7 @@
 from cloudinary_cli.core.config import config
 from cloudinary_cli.core.search import search
 from cloudinary_cli.core.uploader import uploader
-from cloudinary_cli.core.utils import url
+from cloudinary_cli.core.utils import url, utils
 from cloudinary_cli.core.overrides import resolve_command
 
 setattr(click.MultiCommand, "resolve_command", resolve_command)
@@ -15,4 +15,5 @@
     admin,
     uploader,
     url,
+    utils
 ]

cloudinary_cli/core/overrides.py

@@ -2,6 +2,7 @@
 from click.utils import make_str
 from cloudinary import api, uploader
 from cloudinary.uploader import upload as original_upload
+from cloudinary.utils import cloudinary_url as original_cloudinary_url
 
 
 # overrides click.MultiCommand.resolve_command
@@ -37,3 +38,8 @@ def upload(file, **options):
     if "resource_type" not in options.keys():
         options["resource_type"] = "auto"
     return original_upload(file, **options)
+
+
+# Patch to return only the URL
+def cloudinary_url(source, **options):
+    return original_cloudinary_url(source, **options)[0]

cloudinary_cli/core/utils.py

@@ -1,7 +1,29 @@
 from webbrowser import open as open_url
 
 from click import command, argument, option, Choice, echo
-from cloudinary.utils import cloudinary_url
+from cloudinary import utils as cld_utils
+
+from cloudinary_cli.core.overrides import cloudinary_url
+from cloudinary_cli.utils.api_utils import handle_command
+from cloudinary_cli.utils.utils import print_help
+
+cld_utils.cloudinary_url = cloudinary_url
+
+utils_list = ["api_sign_request", "cloudinary_url", "download_archive_url", "download_zip_url",
+              "download_backedup_asset", "verify_api_response_signature", "verify_notification_signature"]
+
+
+@command("utils", help="Call Cloudinary utility methods.")
+@argument("params", nargs=-1)
+@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("-ls", "--ls", is_flag=True, help="List all available utility methods.")
+def utils(params, optional_parameter, optional_parameter_parsed, ls):
+    if ls or len(params) < 1:
+        return print_help(cld_utils, allow_list=utils_list)
+
+    echo(handle_command(params, optional_parameter, optional_parameter_parsed, cld_utils, "Utils"))
 
 
 @command("url", help="Generate a Cloudinary URL, which you can optionally open in your browser.")
@@ -20,7 +42,7 @@ def url(public_id, transformation, resource_type, delivery_type, open_in_browser
         public_id += ".json"
 
     res = cloudinary_url(public_id, resource_type=resource_type,
-                         raw_transformation=transformation, type=delivery_type, sign_url=sign)[0]
+                         raw_transformation=transformation, type=delivery_type, sign_url=sign)
     echo(res)
 
     if open_in_browser:

cloudinary_cli/utils/api_utils.py

@@ -68,6 +68,31 @@ def download_file(remote_file, local_path):
     logger.info(style("Downloaded '{}' to '{}'".format(remote_file['relative_path'], local_path), fg="green"))
 
 
+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)
+
+
 def handle_api_command(
         params,
         optional_parameter,
@@ -87,20 +112,12 @@ def handle_api_command(
     if ls or len(params) < 1:
         return print_help(api_instance)
 
-    try:
-        func = api_instance.__dict__[params[0]]
-        if not callable(func):
-            raise Exception(f"{func} is not callable")
-    except Exception:
-        raise Exception(f"Method {params[0]} does not exist in the {api_name.capitalize()} API.")
-
-    args, kwargs = parse_args_kwargs(func, params[1:]) if len(params) > 1 else ([], {})
-
-    res = func(*args, **{
-        **kwargs,
-        **{k: v for k, v in optional_parameter},
-        **{k: parse_option_value(v) for k, v in optional_parameter_parsed},
-    })
+    res = handle_command(
+        params,
+        optional_parameter,
+        optional_parameter_parsed,
+        api_instance,
+        api_name)
 
     print_json(res)
 

cloudinary_cli/utils/utils.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-
+import builtins
 import json
 import os
 from csv import DictWriter
@@ -13,7 +13,7 @@
 
 from cloudinary_cli.defaults import logger, TEMPLATE_FOLDER
 
-not_callable = ['is_appengine_sandbox', 'call_tags_api', 'call_context_api', 'call_cacheable_api', 'call_api', 'text']
+not_callable = ('is_appengine_sandbox', 'call_tags_api', 'call_context_api', 'call_cacheable_api', 'call_api', 'text')
 
 BLOCK_SIZE = 65536
 
@@ -29,14 +29,27 @@ def etag(fi):
     return file_hash.hexdigest()
 
 
-def get_help_str(api):
-    funcs = list(filter(lambda x: callable(api.__dict__[x]) and x[0].islower() and x not in not_callable,
-                        api.__dict__.keys()))
-    return '\n'.join(["{0:25}({1})".format(x, ", ".join(list(signature(api.__dict__[x]).parameters))) for x in funcs])
+def is_builtin_class_instance(obj):
+    return type(obj).__name__ in dir(builtins)
+
+
+def get_help_str(module, block_list=(), allow_list=()):
+    funcs = list(filter(
+        lambda f:
+        callable(module.__dict__[f])
+        and not is_builtin_class_instance(module.__dict__[f])
+        and f[0].islower()
+        and (f not in block_list and block_list)
+        and (f in allow_list or not allow_list),
+        module.__dict__.keys()))
+
+    template = "{0:" + str(len(max(funcs, key=len)) + 1) + "}({1})"  # Gets the maximal length of the functions' names
+
+    return '\n'.join([template.format(f, ", ".join(list(signature(module.__dict__[f]).parameters))) for f in funcs])
 
 
-def print_help(api):
-    logger.info(get_help_str(api))
+def print_help(api, block_list=not_callable, allow_list=()):
+    logger.info(get_help_str(api, block_list=block_list, allow_list=allow_list))
 
 
 def log_exception(e, message=None):