Clean up and refactor code

* Refactor admin and upload api command handling

* Make `cli.py` executable

* Refactor, extract and reorganise methods

Improve readability

* Fix logging level

Eliminate `thread_utils`

* Add deletion confirmation for `sync` command

Author: const-cloudinary

cloudinary_cli/__init__.py

@@ -1,5 +1,5 @@
-import cloudinary
-
 __version__ = "0.4.2"
 
-cloudinary.USER_PLATFORM = "CloudinaryCLI/{}".format(__version__)
+import cloudinary
+
+cloudinary.USER_PLATFORM = f"CloudinaryCLI/{__version__}"

cloudinary_cli/cli.py

@@ -1,44 +1,66 @@
 #!/usr/bin/env python3
-import json
-import os
+
 import click
+import click_log
 import cloudinary
+
 import cloudinary_cli.core
 import cloudinary_cli.modules
 import cloudinary_cli.samples
-import click_log
-
-from cloudinary_cli.defaults import CLOUDINARY_CLI_CONFIG_FILE
-from cloudinary_cli.utils import logger, initialize
+from cloudinary_cli.defaults import logger
+from cloudinary_cli.utils.config_utils import initialize, load_config, refresh_cloudinary_config
+from cloudinary_cli.utils.utils import log_exception
 
 CONTEXT_SETTINGS = dict(max_content_width=click.get_terminal_size()[0], terminal_width=click.get_terminal_size()[0])
 
 
 @click.group(context_settings=CONTEXT_SETTINGS)
-@click.option("-c", "--config", help="""Tell the CLI which account to run the command on by specifying an account environment variable. 
-""")
-@click.option("-C", "--config_saved", help="""Tell the CLI which account to run the command on by specifying a saved configuration - see `config` command.""")
+@click.option("-c", "--config",
+              help="""Tell the CLI which account to run the command on by specifying an account environment variable."""
+              )
+@click.option("-C", "--config_saved",
+              help="""Tell the CLI which account to run the command on by specifying a saved configuration - see 
+              `config` command.""")
 @click_log.simple_verbosity_option(logger)
 def cli(config, config_saved):
     if config:
-        os.environ.update(dict(CLOUDINARY_URL=config))
+        refresh_cloudinary_config(config)
     elif config_saved:
-        with open(CLOUDINARY_CLI_CONFIG_FILE) as f:
-            os.environ.update(dict(CLOUDINARY_URL=json.loads(f.read())[config_saved]))
-    cloudinary.reset_config()
+        config = load_config()
+        if config_saved not in config:
+            raise Exception(f"Config {config_saved} does not exist")
+
+        refresh_cloudinary_config(config[config_saved])
+
     if cloudinary.config().cloud_name is None:
         logger.warning("No Cloudinary configuration found.")
-    pass
 
+    return 0
 
-cloudinary_cli.core.import_commands(cli)
-cloudinary_cli.modules.import_commands(cli)
-cloudinary_cli.samples.import_commands(cli)
+
+def import_commands(*command_modules):
+    for command_module in command_modules:
+        for command in command_module:
+            cli.add_command(command)
+
+
+import_commands(
+    cloudinary_cli.core.commands,
+    cloudinary_cli.modules.commands,
+    cloudinary_cli.samples.commands,
+)
 
 
 def main():
     initialize()
     try:
-        cli()
+        exit_status = cli()
     except Exception as e:
-        logger.error(str(e))
+        log_exception(e, "Command execution failed")
+        exit_status = 1
+
+    return exit_status
+
+
+if __name__ == "__main__":
+    main()

cloudinary_cli/core/__init__.py

@@ -1,21 +1,18 @@
 import click
-import cloudinary.uploader
 
 from cloudinary_cli.core.admin import admin
 from cloudinary_cli.core.config import config
-from cloudinary_cli.core.overrides import resolve_command, upload
 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.overrides import resolve_command
 
 setattr(click.MultiCommand, "resolve_command", resolve_command)
-cloudinary.uploader.upload = upload
 
-
-def import_commands(cli):
-
-    cli.add_command(config)
-    cli.add_command(search)
-    cli.add_command(admin)
-    cli.add_command(uploader)
-    cli.add_command(url)
+commands = [
+    config,
+    search,
+    admin,
+    uploader,
+    url,
+]

cloudinary_cli/core/admin.py

@@ -1,9 +1,7 @@
-from webbrowser import open as open_url
-
 from click import command, argument, option
 from cloudinary import api
 
-from cloudinary_cli.utils import get_help, parse_args_kwargs, parse_option_value, log_json, write_out, logger
+from cloudinary_cli.utils.api_utils import handle_api_command
 
 
 @command("admin",
@@ -25,24 +23,7 @@
 @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):
-    if doc:
-        open_url("https://cloudinary.com/documentation/admin_api")
-        return
-    if ls or len(params) < 1:
-        logger.info(get_help(api))
-        return
-    try:
-        func = api.__dict__[params[0]]
-        if not callable(func):
-            raise Exception("{} is not callable".format(func))
-    except Exception:
-        raise Exception("Method {} does not exist in the Admin API.".format(params[0]))
-    parameters, options = parse_args_kwargs(func, params[1:]) if len(params) > 1 else ([], {})
-    res = func(*parameters, **{
-        **options,
-        **{k: v for k, v in optional_parameter},
-        **{k: parse_option_value(v) for k, v in optional_parameter_parsed},
-    })
-    log_json(res)
-    if save:
-        write_out(res, save)
+    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")

cloudinary_cli/core/config.py

@@ -1,64 +1,40 @@
-import json
-
 import cloudinary
-from click import command, option
-from cloudinary import api
+from click import command, option, echo
 
-from cloudinary_cli.utils import CLOUDINARY_CLI_CONFIG_FILE, refresh_config, logger
+from cloudinary_cli.defaults import logger
+from cloudinary_cli.utils.config_utils import load_config, verify_cloudinary_url, update_config, remove_config_keys
 
 
 @command("config", help="Display the current configuration, and manage additional configurations.")
 @option("-n", "--new", help="""\b Create and name a configuration from a Cloudinary account environment variable.
 e.g. cld config -n <NAME> <CLOUDINARY_URL>""", nargs=2)
 @option("-ls", "--ls", help="List all saved configurations.", is_flag=True)
 @option("-rm", "--rm", help="Delete a specified configuration.", nargs=1)
-@option("-url", "--from_url", help="Create a configuration from a Cloudinary account environment variable. The configuration name is the cloud name.", nargs=1)
+@option("-url", "--from_url",
+        help="Create a configuration from a Cloudinary account environment variable. "
+             "The configuration name is the cloud name.",
+        nargs=1)
 def config(new, ls, rm, from_url):
-    if not (new or ls or rm or from_url):
-        logger.info('\n'.join(["{}:\t{}".format(k, v if k != "api_secret"
-        else "***************{}".format(v[-4:]))
-                               for k, v in cloudinary.config().__dict__.items()]))
-        return
+    if new or from_url:
+        config_name, cloudinary_url = new or [None, from_url]
 
-    with open(CLOUDINARY_CLI_CONFIG_FILE, "r+") as f:
-        fi = f.read()
-        cfg = json.loads(fi) if fi != "" else {}
-        f.close()
-    if new:
-        try:
-            refresh_config(new[1])
-            cfg[new[0]] = new[1]
-            api.ping()
-            with open(CLOUDINARY_CLI_CONFIG_FILE, "w") as f:
-                f.write(json.dumps(cfg))
-                f.close()
-            logger.info("Config '{}' saved!".format(new[0]))
-        except Exception as e:
-            logger.error("Invalid Cloudinary URL: {}".format(new[1]))
-            raise e
-        return
-    if ls:
-        logger.info("\n".join(cfg.keys()))
-    if rm:
-        if rm not in cfg.keys():
-            logger.warn("Configuration '{}' not found.".format(rm))
+        if not verify_cloudinary_url(cloudinary_url):
             return
-        del cfg[rm]
-        open(CLOUDINARY_CLI_CONFIG_FILE, "w").write(json.dumps(cfg))
-        logger.info("Configuration '{}' deleted".format(rm))
-        return
-    if from_url:
-        if "CLOUDINARY_URL=" in from_url:
-            from_url = from_url[15:]
-        try:
-            refresh_config(from_url)
-            cfg[cloudinary.config().cloud_name] = from_url
-            api.ping()
-            with open(CLOUDINARY_CLI_CONFIG_FILE, "w") as f:
-                f.write(json.dumps(cfg))
-                f.close()
-            logger.info("Config '{}' saved!".format(cloudinary.config().cloud_name))
-            logger.info("Example usage: cld -C {} <command>".format(cloudinary.config().cloud_name))
-        except Exception as e:
-            logger.error("Invalid Cloudinary URL: {}".format(from_url))
-            raise e
+
+        config_name = config_name or cloudinary.config().cloud_name
+
+        update_config({config_name: cloudinary_url})
+
+        logger.info("Config '{}' saved!".format(config_name))
+        logger.info("Example usage: cld -C {} <command>".format(config_name))
+    elif rm:
+        if remove_config_keys(rm):
+            logger.warn(f"Configuration '{rm}' not found.")
+        else:
+            logger.info(f"Configuration '{rm}' deleted")
+    elif ls:
+        echo("\n".join(load_config().keys()))
+    else:
+        obfuscated_config = {k: v if k != "api_secret" else "***************{}".format(v[-4:])
+                             for k, v in cloudinary.config().__dict__.items()}
+        echo('\n'.join(["{}:\t{}".format(k, v) for k, v in obfuscated_config.items()]))