Refactor code, normalize parameters

Author: const-cloudinary

cloudinary_cli/modules/clone.py

@@ -1,103 +1,87 @@
 from click import command, option, style
-from cloudinary_cli.utils.utils import normalize_list_params, \
-     print_help_and_exit
+from cloudinary_cli.utils.utils import normalize_list_params, print_help_and_exit
 import cloudinary
 from cloudinary_cli.utils.utils import run_tasks_concurrently
 from cloudinary_cli.utils.api_utils import upload_file
-from cloudinary_cli.utils.config_utils import load_config
+from cloudinary_cli.utils.config_utils import load_config, get_cloudinary_config, config_to_dict
 from cloudinary_cli.defaults import logger
-from cloudinary_cli.core.search import execute_single_request, \
-     handle_auto_pagination
+from cloudinary_cli.core.search import execute_single_request, handle_auto_pagination
 
 DEFAULT_MAX_RESULTS = 500
 
 
 @command("clone",
-         short_help="""Clone assets from one account to another.""",
+         short_help="""Clone assets from one product environment to another.""",
          help="""
 \b
-Clone assets from one environment to another with/without tags and/or context (structured metadata is not currently supported).
+Clone assets from one product environment to another with/without tags and/or context (structured metadata is not currently supported).
 Source will be your `CLOUDINARY_URL` environemnt variable but you also can specify a different source using `-c/-C` option.
 Cloning restricted assets is also not supported currently.
 Format: cld clone -T <target_environment> <command options>
 `<target_environment>` can be a CLOUDINARY_URL or a saved config (see  `config` command)
 e.g. cld clone -T cloudinary://<api_key>:<api_secret>@<cloudname> -f tags,context
 """)
-@option("-T", "--target",
+@option("-t", "--target",
         help="Tell the CLI the target environemnt to run the command on.")
-@option("-F", "--force", is_flag=True,
+@option("-f", "--force", is_flag=True,
         help="Skip confirmation.")
-@option("-O", "--overwrite", is_flag=True, default=False,
+@option("-o", "--overwrite", is_flag=True, default=False,
         help="Specify whether to overwrite existing assets.")
 @option("-w", "--concurrent_workers", type=int, default=30,
         help="Specify the number of concurrent network threads.")
-@option("-f", "--fields", multiple=True,
+@option("-fi", "--fields", multiple=True,
         help="Specify whether to copy tags and context.")
 @option("-se", "--search_exp", default="",
         help="Define a search expression.")
 @option("--async", "async_", is_flag=True, default=False,
         help="Generate asynchronously.")
 @option("-nu", "--notification_url",
         help="Webhook notification URL.")
-def clone(target, force, overwrite, concurrent_workers, fields, search_exp,
-          async_, notification_url):
+def clone(target, force, overwrite, concurrent_workers, fields, search_exp, async_, notification_url):
     if not target:
         print_help_and_exit()
 
-    target_config = cloudinary.Config()
-    is_cloudinary_url = False
-    if target.startswith("cloudinary://"):
-        is_cloudinary_url = True
-        parsed_url = target_config._parse_cloudinary_url(target)
-    elif target in load_config():
-        parsed_url = target_config._parse_cloudinary_url(load_config().get(target))
-    else:
-        logger.error("The specified config does not exist or the "
-                     "CLOUDINARY_URL scheme provided is invalid "
-                     "(expecting to start with 'cloudinary://').")
+    target_config = get_cloudinary_config(target)
+    if not target_config:
+        logger.error("The specified config does not exist or the CLOUDINARY_URL scheme provided is invalid"
+                     " (expecting to start with 'cloudinary://').")
         return False
 
-    target_config._setup_from_parsed_url(parsed_url)
-    target_config_dict = {k: v for k, v in target_config.__dict__.items()
-                          if not k.startswith("_")}
-    if is_cloudinary_url:
-        try:
-            cloudinary.api.ping(**target_config_dict)
-        except Exception as e:
-            logger.error(f"{e}. Please double-check your Cloudinary URL.")
-            return False
-
-    source_cloudname = cloudinary.config().cloud_name
-    target_cloudname = target_config.cloud_name
-    if source_cloudname == target_cloudname:
-        logger.info("Target environment cannot be the "
-                    "same as source environment.")
-        return True
-
-    copy_fields = normalize_list_params(fields)
-    search = cloudinary.search.Search().expression(search_exp)
-    search.fields(['tags', 'context', 'access_control',
-                   'secure_url', 'display_name'])
-    search.max_results(DEFAULT_MAX_RESULTS)
-    res = execute_single_request(search, fields_to_keep="")
-    res = handle_auto_pagination(res, search, force, fields_to_keep="")
+    if cloudinary.config().cloud_name == target_config.cloud_name:
+        logger.error("Target environment cannot be the same as source environment.")
+        return False
+
+    source_assets = search_assets(force, search_exp)
 
     upload_list = []
-    for r in res.get('resources'):
-        updated_options, asset_url = process_metadata(r, overwrite, async_,
-                                                      notification_url,
-                                                      copy_fields)
-        updated_options.update(target_config_dict)
+    for r in source_assets.get('resources'):
+        updated_options, asset_url = process_metadata(r, overwrite, async_, notification_url,
+                                                      normalize_list_params(fields))
+        updated_options.update(config_to_dict(target_config))
         upload_list.append((asset_url, {**updated_options}))
 
-    logger.info(style(f'Copying {len(upload_list)} asset(s) to '
-                      f'{target_cloudname}', fg="blue"))
-    run_tasks_concurrently(upload_file, upload_list,
-                           concurrent_workers)
+    if not upload_list:
+        logger.error(style(f'No assets found in {cloudinary.config().cloud_name}', fg="red"))
+        return False
+
+    logger.info(style(f'Copying {len(upload_list)} asset(s) from {cloudinary.config().cloud_name} to {target_config.cloud_name}', fg="blue"))
+
+    run_tasks_concurrently(upload_file, upload_list, concurrent_workers)
 
     return True
 
 
+def search_assets(force, search_exp):
+    search = cloudinary.search.Search().expression(search_exp)
+    search.fields(['tags', 'context', 'access_control', 'secure_url', 'display_name'])
+    search.max_results(DEFAULT_MAX_RESULTS)
+
+    res = execute_single_request(search, fields_to_keep="")
+    res = handle_auto_pagination(res, search, force, fields_to_keep="")
+
+    return res
+
+
 def process_metadata(res, overwrite, async_, notification_url, copy_fields=""):
     cloned_options = {}
     asset_url = res.get('secure_url')

cloudinary_cli/utils/config_utils.py

@@ -44,16 +44,31 @@ def refresh_cloudinary_config(cloudinary_url):
 
 def verify_cloudinary_url(cloudinary_url):
     refresh_cloudinary_config(cloudinary_url)
-    try:
-        api.ping()
-    except Exception as e:
-        log_exception(e, f"Invalid Cloudinary URL: {cloudinary_url}")
+    return ping_cloudinary()
+
+
+def get_cloudinary_config(target):
+    target_config = cloudinary.Config()
+    if target.startswith("cloudinary://"):
+        parsed_url = target_config._parse_cloudinary_url(target)
+    elif target in load_config():
+        parsed_url = target_config._parse_cloudinary_url(load_config().get(target))
+    else:
         return False
-    return True
 
+    target_config._setup_from_parsed_url(parsed_url)
+
+    if not ping_cloudinary(**config_to_dict(target_config)):
+        logger.error(f"Invalid Cloudinary config: {target}")
+        return False
+
+    return target_config
+
+def config_to_dict(config):
+    return {k: v for k, v in config.__dict__.items() if not k.startswith("_")}
 
 def show_cloudinary_config(cloudinary_config):
-    obfuscated_config = {k: v for k, v in cloudinary_config.__dict__.items() if not k.startswith("_")}
+    obfuscated_config = config_to_dict(cloudinary_config)
 
     if "api_secret" in obfuscated_config:
         api_secret = obfuscated_config["api_secret"]
@@ -96,6 +111,14 @@ def is_valid_cloudinary_config():
 def initialize():
     migrate_old_config()
 
+def ping_cloudinary(**options):
+    try:
+        api.ping(**options)
+    except Exception as e:
+        logger.error(f"Failed to ping Cloudinary: {e}")
+        return False
+
+    return True
 
 def _verify_file_path(file):
     os.makedirs(os.path.dirname(file), exist_ok=True)