small refactoring :)

Author: const-cloudinary

cloudinary_cli/modules/clone.py

@@ -11,6 +11,7 @@
 import re
 
 DEFAULT_MAX_RESULTS = 500
+ALLOWED_TYPE_VALUES = ("upload", "private", "authenticated")
 
 
 @command("clone",
@@ -49,6 +50,30 @@
               "bandwidth costs."))
 def clone(target, force, overwrite, concurrent_workers, fields,
           search_exp, async_, notification_url, ttl):
+    target_config, auth_token = _validate_clone_inputs(target)
+    if not target_config:
+        return False
+
+    source_assets = search_assets(search_exp, force)
+    if not source_assets or not source_assets.get('resources'):
+        logger.error(style(f"No asset(s) found in {cloudinary.config().cloud_name}", fg="red"))
+        return False
+
+    upload_list = _prepare_upload_list(
+        source_assets, target_config, overwrite, async_,
+        notification_url, auth_token, ttl, fields
+    )
+
+    logger.info(style(f"Copying {len(upload_list)} asset(s) from "
+                      f"{cloudinary.config().cloud_name} to "
+                      f"{target_config.cloud_name}", fg="blue"))
+
+    run_tasks_concurrently(upload_file, upload_list, concurrent_workers)
+
+    return True
+
+
+def _validate_clone_inputs(target):
     if not target:
         print_help_and_exit()
 
@@ -57,12 +82,12 @@ def clone(target, force, overwrite, concurrent_workers, fields,
         logger.error("The specified config does not exist or the "
                      "CLOUDINARY_URL scheme provided is invalid "
                      "(expecting to start with 'cloudinary://').")
-        return False
+        return None, None
 
     if cloudinary.config().cloud_name == target_config.cloud_name:
         logger.error("Target environment cannot be the same "
                      "as source environment.")
-        return False
+        return None, None
 
     auth_token = cloudinary.config().auth_token
     if auth_token:
@@ -74,13 +99,13 @@ def clone(target, force, overwrite, concurrent_workers, fields,
         except Exception as e:
             logger.error(f"{e} - auth_token validation failed. "
                          "Please double-check your auth_token parameters.")
-            return False
+            return None, None
+
+    return target_config, auth_token
 
-    source_assets = search_assets(force, search_exp)
-    if not source_assets:
-        # End command if search_exp contains unsupported type(s)
-        return False
 
+def _prepare_upload_list(source_assets, target_config, overwrite, async_,
+                         notification_url, auth_token, ttl, fields):
     upload_list = []
     for r in source_assets.get('resources'):
         updated_options, asset_url = process_metadata(r, overwrite, async_,
@@ -89,42 +114,13 @@ def clone(target, force, overwrite, concurrent_workers, fields,
                                                       normalize_list_params(fields))
         updated_options.update(config_to_dict(target_config))
         upload_list.append((asset_url, {**updated_options}))
+    return upload_list
 
-    source_cloud_name = cloudinary.config().cloud_name
-    if not upload_list:
-        logger.error(style('No asset(s) found in '
-                           f'{source_cloud_name}', fg="red"))
-        return False
-
-    logger.info(style(f'Copying {len(upload_list)} asset(s) from '
-                      f'{source_cloud_name} to '
-                      f'{target_config.cloud_name}', fg="blue"))
-
-    run_tasks_concurrently(upload_file, upload_list, concurrent_workers)
-
-    return True
 
-
-def search_assets(force, search_exp):
-    # Prevent other unsupported types to prevent
-    # avoidable errors during the upload process
-    # and append the default types in not in the
-    # search expression
-    ALLOWED_TYPES = {"type:upload", "type:private", "type:authenticated",
-                     "type=upload", "type=private", "type=authenticated"}
-    if search_exp and re.search(r"\btype\s*[:=]\s*\w+", search_exp):
-        exp_types = re.findall(r"\btype\s*[:=]\s*\w+", search_exp)
-        exp_types_cleaned = [''.join(t.split()) for t in exp_types]
-        unallowed_types = [t for t in exp_types_cleaned if t not in ALLOWED_TYPES]
-        if unallowed_types:
-            logger.error("Unsupported type(s) in search expression: "
-                         f"{', '.join(unallowed_types)}. "
-                         "Only upload/private/authenticated types allowed.")
-            return False
-    elif search_exp:
-        search_exp += " AND (type:upload OR type:private OR type:authenticated)"
-    else:
-        search_exp = "type:upload OR type:private OR type:authenticated"
+def search_assets(search_exp, force):
+    search_exp = _normalize_search_expression(search_exp)
+    if not search_exp:
+        return False
 
     search = cloudinary.search.Search().expression(search_exp)
     search.fields(['tags', 'context', 'access_control',
@@ -137,64 +133,110 @@ def search_assets(force, search_exp):
     return res
 
 
-def process_metadata(res, overwrite, async_, notification_url,
-                     auth_token, ttl, copy_fields=""):
-    cloned_options = {}
-    acc_ctl = res.get('access_control')
-    pub_id = res.get('public_id')
-    del_type = res.get('type')
+def _normalize_search_expression(search_exp):
+    """
+    Ensures the search expression has a valid 'type' filter.
+
+    - If no expression is given, a default is created.
+    - If 'type' filters exist, they are validated.
+    - If no 'type' filters exist, the default is appended.
+    """
+    default_types_str = " OR ".join(f"type:{t}" for t in ALLOWED_TYPE_VALUES)
+
+    if not search_exp:
+        return default_types_str
+
+    # Use a simple regex to find all 'type' filters
+    found_types = re.findall(r"\btype\s*[:=]\s*(\w+)", search_exp)
+
+    if not found_types:
+        # No 'type' filter found, so append the default
+        return f"{search_exp} AND ({default_types_str})"
+
+    # A 'type' filter was found, so validate it
+    invalid_types = {t for t in found_types if t not in ALLOWED_TYPE_VALUES}
+
+    if invalid_types:
+        error_msg = ", ".join(f"type:{t}" for t in invalid_types)
+        logger.error(
+            f"Unsupported type(s) in search expression: {error_msg}. "
+            f"Only {', '.join(ALLOWED_TYPE_VALUES)} types allowed."
+        )
+        return None
+
+    # All found types are valid, so return the original expression
+    return search_exp
+
+
+def process_metadata(res, overwrite, async_, notification_url, auth_token, ttl, copy_fields=None):
+    if copy_fields is None:
+        copy_fields = []
+    asset_url = _get_asset_url(res, auth_token, ttl)
+    cloned_options = _build_cloned_options(res, overwrite, async_, notification_url, copy_fields)
+
+    return cloned_options, asset_url
+
+
+def _get_asset_url(res, auth_token, ttl):
+    if not (isinstance(res.get('access_control'), list) and
+            len(res.get('access_control')) > 0 and
+            isinstance(res['access_control'][0], dict) and
+            res['access_control'][0].get("access_type") == "token"):
+        return res.get('secure_url')
+
     reso_type = res.get('resource_type')
+    del_type = res.get('type')
+    pub_id = res.get('public_id')
     file_format = res.get('format')
-    if (
-        isinstance(acc_ctl, list)
-        and len(acc_ctl) > 0
-        and isinstance(acc_ctl[0], dict)
-        and acc_ctl[0].get("access_type") == "token"
-    ):
-        # Generate a time-limited URL for restricted assets
-        # Use private url if no auth_token provided
-        if auth_token:
-            # Don't add format if asset is raw
-            pub_id_format = (pub_id if reso_type == "raw"
-                             else f"{pub_id}.{file_format}")
-            asset_url = cloudinary.utils.cloudinary_url(
-                            pub_id_format,
-                            type=del_type,
-                            resource_type=reso_type,
-                            auth_token={"duration": ttl},
-                            secure=True,
-                            sign_url=True)
-        else:
-            expiry_date = int(time.time()) + ttl
-            asset_url = cloudinary.utils.private_download_url(
-                            pub_id,
-                            file_format,
-                            resource_type=reso_type,
-                            type=del_type,
-                            expires_at=expiry_date)
-    else:
-        asset_url = res.get('secure_url')
-    cloned_options['access_control'] = acc_ctl
-    cloned_options['public_id'] = pub_id
-    cloned_options['type'] = del_type
-    cloned_options['resource_type'] = reso_type
-    cloned_options['overwrite'] = overwrite
-    cloned_options['async'] = async_
-    if "tags" in copy_fields:
-        cloned_options['tags'] = res.get('tags')
-    if "context" in copy_fields:
-        cloned_options['context'] = res.get('context')
+
+    if auth_token:
+        # Raw assets already have the format in the public_id
+        pub_id_format = pub_id if reso_type == "raw" else f"{pub_id}.{file_format}"
+        return cloudinary.utils.cloudinary_url(
+            pub_id_format,
+            type=del_type,
+            resource_type=reso_type,
+            auth_token={"duration": ttl},
+            secure=True,
+            sign_url=True
+        )
+
+    # Use private url if no auth_token provided
+    return cloudinary.utils.private_download_url(
+        pub_id,
+        file_format,
+        resource_type=reso_type,
+        type=del_type,
+        expires_at=int(time.time()) + ttl
+    )
+
+
+def _build_cloned_options(res, overwrite, async_, notification_url, copy_fields):
+    # 1. Start with mandatory options
+    cloned_options = {
+        'overwrite': overwrite,
+        'async': async_,
+    }
+
+    # 2. Copy fields from source asset. Some are standard, others are from user input.
+    fields_to_copy = {'public_id', 'type', 'resource_type', 'access_control'}.union(copy_fields)
+    cloned_options.update({field: res.get(field) for field in fields_to_copy})
+
+    # 3. Handle fields that are added only if they have a truthy value
+    if res.get('display_name'):
+        cloned_options['display_name'] = res['display_name']
+
+    # This is required to put the asset in the correct asset_folder
+    # when copying from a fixed to DF (dynamic folder) cloud as if
+    # you just pass a `folder` param to a DF cloud, it will append
+    # this to the `public_id` and we don't want this.
     if res.get('folder'):
-        # This is required to put the asset in the correct asset_folder
-        # when copying from a fixed to DF (dynamic folder) cloud as if
-        # you just pass a `folder` param to a DF cloud, it will append
-        # this to the `public_id` and we don't want this.
-        cloned_options['asset_folder'] = res.get('folder')
+        cloned_options['asset_folder'] = res['folder']
     elif res.get('asset_folder'):
-        cloned_options['asset_folder'] = res.get('asset_folder')
-    if res.get('display_name'):
-        cloned_options['display_name'] = res.get('display_name')
+        cloned_options['asset_folder'] = res['asset_folder']
+
     if notification_url:
         cloned_options['notification_url'] = notification_url
 
-    return cloned_options, asset_url
+    # 4. Clean up any None values before returning
+    return {k: v for k, v in cloned_options.items() if v is not None}

test/test_modules/test_cli_clone.py

@@ -1,14 +1,12 @@
 import unittest
 from unittest.mock import patch, MagicMock
 import re
-import importlib.util
-import os
-
-# Get the path to the clone module and load it directly to avoid conflicts with the command object
-clone_module_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', 'cloudinary_cli', 'modules', 'clone.py'))
-spec = importlib.util.spec_from_file_location("clone_module", clone_module_path)
-clone_module = importlib.util.module_from_spec(spec)
-spec.loader.exec_module(clone_module)
+import sys
+
+# Import the modules package, which will load the clone module.
+# The 'clone' name in the package is the command object, so we get the module from sys.modules.
+import cloudinary_cli.modules
+clone_module = sys.modules['cloudinary_cli.modules.clone']
 
 from cloudinary_cli.defaults import logger