Improve error handling of CLI commands

Author: const-cloudinary

cloudinary_cli/cli.py

@@ -1,51 +1,14 @@
 #!/usr/bin/env python3
-import platform
-import shutil
 import sys
 
-import click
-import click_log
-import cloudinary
+from click import ClickException
 
 import cloudinary_cli.core
 import cloudinary_cli.modules
 import cloudinary_cli.samples
-from cloudinary_cli.defaults import logger
-from cloudinary_cli.utils.config_utils import initialize, load_config, refresh_cloudinary_config, \
-    is_valid_cloudinary_config
+from cloudinary_cli.cli_group import cli
+from cloudinary_cli.utils.config_utils import initialize
 from cloudinary_cli.utils.utils import log_exception, ConfigurationError
-from cloudinary_cli.version import __version__ as cli_version
-
-CONTEXT_SETTINGS = dict(max_content_width=shutil.get_terminal_size()[0], terminal_width=shutil.get_terminal_size()[0])
-
-
-@click.group(context_settings=CONTEXT_SETTINGS)
-@click.help_option()
-@click.version_option(cli_version, prog_name="Cloudinary CLI",
-                      message=f"%(prog)s, version %(version)s\n"
-                              f"Cloudinary SDK, version {cloudinary.VERSION}\n"
-                              f"Python, version {platform.python_version()}")
-@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:
-        refresh_cloudinary_config(config)
-    elif config_saved:
-        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 not is_valid_cloudinary_config():
-        logger.warning("No Cloudinary configuration found.")
-
-    return 0
 
 
 def import_commands(*command_modules):
@@ -62,21 +25,26 @@ def import_commands(*command_modules):
 
 
 def main():
+    exit_status = 1  # very optimistic :)
+
     initialize()
+
     try:
-        exit_status = cli()
+        # we don't use standalone mode to get the return value from the command execution
+        exit_status = cli.main(standalone_mode=False)
+    except ClickException as e:
+        # show usage with error message
+        e.show()
     except ConfigurationError as e:
         log_exception(e)
-        exit_status = 1
     except Exception as e:
         # Improve configuration error handling
         if "Must supply cloud_name" in str(e):
             log_exception("No Cloudinary configuration found.")
         else:
             log_exception(e, "Command execution failed")
-        exit_status = 1
 
-    return exit_status
+    return 0 if exit_status or exit_status is None else 1
 
 
 if __name__ == "__main__":

cloudinary_cli/cli_group.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+import platform
+import shutil
+
+import click
+import click_log
+import cloudinary
+
+from cloudinary_cli.defaults import logger
+from cloudinary_cli.utils.config_utils import load_config, refresh_cloudinary_config, \
+    is_valid_cloudinary_config
+from cloudinary_cli.version import __version__ as cli_version
+
+CONTEXT_SETTINGS = dict(max_content_width=shutil.get_terminal_size()[0], terminal_width=shutil.get_terminal_size()[0])
+
+
+@click.group(context_settings=CONTEXT_SETTINGS)
+@click.help_option()
+@click.version_option(cli_version, prog_name="Cloudinary CLI",
+                      message=f"%(prog)s, version %(version)s\n"
+                              f"Cloudinary SDK, version {cloudinary.VERSION}\n"
+                              f"Python, version {platform.python_version()}")
+@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:
+        refresh_cloudinary_config(config)
+    elif config_saved:
+        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 not is_valid_cloudinary_config():
+        logger.warning("No Cloudinary configuration found.")
+
+    return 0

cloudinary_cli/modules/make.py

@@ -1,13 +1,14 @@
 import os
 
-from click import command, argument, echo, option
+from click import argument, echo, option
 
+from cloudinary_cli.cli_group import cli
 from cloudinary_cli.defaults import TEMPLATE_EXTS, TEMPLATE_FOLDER
 from cloudinary_cli.utils.utils import load_template, print_help_and_exit
 
 
-@command("make", short_help="Return template code for implementing the specified Cloudinary widget.",
-         help="""\b
+@cli.command("make", short_help="Return template code for implementing the specified Cloudinary widget.",
+             help="""\b
 Return template code for implementing the specified Cloudinary widget.
 e.g. cld make media library widget
      cld make python find all empty folders
@@ -37,7 +38,14 @@ def make(template, list_languages, list_templates):
                     echo(template_file.name.replace("_", " "))
         return True
 
-    echo(load_template(language, '_'.join(template)))
+    template_result = load_template(language, '_'.join(template))
+
+    if not template_result:
+        return False
+
+    echo(template_result)
+
+    return True
 
 
 def _handle_language_and_template(language_and_template):

cloudinary_cli/modules/migrate.py

@@ -1,11 +1,13 @@
+import os
 from os.path import join
 
 from click import command, argument, option
 from cloudinary import api
+from cloudinary.exceptions import Error
 from cloudinary.utils import cloudinary_url
 from requests import head
 
-from cloudinary_cli.utils.utils import logger
+from cloudinary_cli.utils.utils import logger, log_exception
 
 
 @command("migrate",
@@ -17,14 +19,39 @@
 @option("-d", "--delimiter", default="\n", help="The separator used between the URLs. Default: New line")
 @option("-v", "--verbose", is_flag=True)
 def migrate(upload_mapping, file, delimiter, verbose):
-    with open(file) as f:
-        items = f.read().split(delimiter)
-    mapping = api.upload_mapping(upload_mapping)
-    items = map(lambda x: cloudinary_url(join(mapping['folder'], x[len(mapping['template']):])),
-                filter(lambda x: x != '', items))
-    for i in items:
-        res = head(i)
+    if not os.path.exists(file):
+        logger.error(f"Migration file: '{file}' does not exist")
+        return False
+
+    try:
+        with open(file) as f:
+            migration_files = f.read().split(delimiter)
+    except IOError as e:
+        log_exception(e, f"Failed reading migration file: '{file}'")
+        return False
+
+    try:
+        mapping = api.upload_mapping(upload_mapping)
+    except Error as e:
+        log_exception(e, f"Failed retrieving upload mapping: '{upload_mapping}'")
+        return False
+
+    exit_status = True
+    migration_urls = []
+
+    for migration_file in filter(None, migration_files):  # omit empty lines
+        if not migration_file.startswith(mapping['template']):
+            logger.error(f"Skipping '{migration_file}', it does not belong to the upload mapping")
+            exit_status = False
+            continue
+
+        migration_urls.append(cloudinary_url(join(mapping['folder'], migration_file[len(mapping['template']):])))
+
+    for migration_url in migration_urls:
+        res = head(migration_url)
         if res.status_code != 200:
-            logger.error("Failed uploading asset: " + res.__dict__['headers']['X-Cld-Error'])
+            logger.error(f"Failed uploading {migration_url}: {res.__dict__['headers']['X-Cld-Error']}")
         elif verbose:
-            logger.info("Uploaded {}".format(i))
+            logger.info(f"Uploaded {migration_url}")
+
+    return exit_status

cloudinary_cli/modules/sync.py

@@ -3,7 +3,7 @@
 from itertools import groupby
 from os import path, remove
 
-from click import command, argument, option, style
+from click import command, argument, option, style, UsageError
 from cloudinary import api
 
 from cloudinary_cli.utils.api_utils import query_cld_folder, upload_file, download_file
@@ -35,12 +35,12 @@
 def sync(local_folder, cloudinary_folder, push, pull, include_hidden, concurrent_workers, force, keep_unique,
          deletion_batch_size):
     if push == pull:
-        raise Exception("Please use either the '--push' OR '--pull' options")
+        raise UsageError("Please use either the '--push' OR '--pull' options")
 
     sync_dir = SyncDir(local_folder, cloudinary_folder, include_hidden, concurrent_workers, force, keep_unique,
                        deletion_batch_size)
 
-    result = 0
+    result = True
     if push:
         result = sync_dir.push()
     elif pull:

cloudinary_cli/modules/upload_dir.py

@@ -1,13 +1,12 @@
-import glob
-from os import getcwd, walk
-from os.path import dirname, split, join as path_join, abspath
+from os import getcwd
+from os.path import dirname, join as path_join
 from pathlib import Path
 
 from click import command, argument, option, style
 
 from cloudinary_cli.utils.api_utils import upload_file
-from cloudinary_cli.utils.utils import parse_option_value, logger, run_tasks_concurrently
 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
 
 
 @command("upload_dir", help="""Upload a folder of assets, maintaining the folder structure.""")
@@ -32,7 +31,11 @@ def upload_dir(directory, glob_pattern, include_hidden, optional_parameter, opti
     items, skipped = {}, {}
 
     dir_to_upload = Path(path_join(getcwd(), directory))
-    logger.info("Uploading directory '{}'".format(dir_to_upload))
+    if not dir_to_upload.exists():
+        logger.error(f"Directory: {dir_to_upload} does not exist")
+        return False
+
+    logger.info(f"Uploading directory '{dir_to_upload}'")
     parent = dirname(dir_to_upload)
     options = {
         **{k: v for k, v in optional_parameter},
@@ -58,5 +61,9 @@ def upload_dir(directory, glob_pattern, include_hidden, optional_parameter, opti
     run_tasks_concurrently(upload_file, uploads, concurrent_workers)
 
     logger.info(style("{} resources uploaded".format(len(items)), fg="green"))
+
     if skipped:
         logger.warn("{} items skipped".format(len(skipped)))
+        return False
+
+    return True

cloudinary_cli/utils/utils.py

@@ -75,6 +75,9 @@ def log_exception(e, message=None):
 
 def load_template(language, template_name):
     filepath = os.path.join(TEMPLATE_FOLDER, language, template_name)
+    if not os.path.exists(filepath):
+        logger.error(f"Template: '{template_name}' for language: '{language}' does not exist")
+        return False
     try:
         with open(filepath) as f:
             template = Environment(loader=FileSystemLoader(TEMPLATE_FOLDER)).from_string(f.read())