Refactor path/public_id formatting

Author: Brian Luk

cloudinary_cli/modules/sync.py

@@ -1,14 +1,14 @@
 import logging
 from functools import reduce
 from itertools import product
-from os import sep, remove
+from os import remove
 from os.path import split, join as path_join, abspath
 
 from click import command, argument, option, style
 from cloudinary import api
 
 from cloudinary_cli.utils.api_utils import query_cld_folder, upload_file, download_file
-from cloudinary_cli.utils.file_utils import walk_dir, delete_empty_dirs
+from cloudinary_cli.utils.file_utils import walk_dir, delete_empty_dirs, get_destination_folder
 from cloudinary_cli.utils.json_utils import print_json
 from cloudinary_cli.utils.utils import logger, run_tasks_concurrently, confirm_action
 
@@ -84,7 +84,6 @@ def push(self):
             return False
 
         files_to_push = self.unique_local_file_names | self.out_of_sync_file_names
-
         to_upload = list(filter(lambda x: split(x)[1][0] != ".", files_to_push))
         logger.info(f"Uploading {len(to_upload)} items to Cloudinary folder '{self.remote_dir}'")
 
@@ -96,7 +95,7 @@ def push(self):
         }
         uploads = []
         for file in to_upload:
-            folder = "/".join([self.remote_dir, *split(file)[:-1]])
+            folder = get_destination_folder(self.remote_dir, file)
 
             uploads.append((self.local_files[file]['path'], {**options, 'folder': folder}))
 

cloudinary_cli/modules/upload_dir.py

@@ -5,6 +5,7 @@
 
 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
 
 
 @command("upload_dir", help="""Upload a folder of assets, maintaining the folder structure.""")
@@ -43,10 +44,11 @@ def upload_dir(directory, optional_parameter, optional_parameter_parsed, transfo
     for root, _, files in walk(dir_to_upload):
         for fi in files:
             file_path = abspath(path_join(dir_to_upload, root, fi))
-            mod_folder = path_join(folder, dirname(file_path[len(parent) + 1:]))
+
             if split(file_path)[1][0] == ".":
                 continue
-            options = {**options, "folder": mod_folder}
+
+            options = {**options, "folder": get_destination_folder(folder, file_path, parent=parent)}
             uploads.append((file_path, options, items, skipped))
 
     run_tasks_concurrently(upload_file, uploads, concurrent_workers)

cloudinary_cli/utils/file_utils.py

@@ -1,4 +1,5 @@
-from os import walk, path, listdir, rmdir
+from os import walk, path, listdir, rmdir, sep
+from os.path import split, relpath, abspath
 
 from cloudinary_cli.defaults import logger
 from cloudinary_cli.utils.utils import etag
@@ -30,3 +31,22 @@ def delete_empty_dirs(root, remove_root=False):
     if len(files) == 0 and remove_root:
         logger.debug(f"Removing empty folder '{root}'")
         rmdir(root)
+
+
+def get_destination_folder(cloudinary_folder: str, file_path: str, parent: str = None) -> str:
+    """
+    :param cloudinary_folder:   Destination folder in Cloudinary
+    :param file_path:           Path to the local file
+    :param parent:              Parent folder of the directory to upload/sync
+    :return:                    Value passed to `folder` when uploading to Cloudinary
+    """
+    folder_path = []
+
+    parent_path = abspath(parent) if parent else None
+    splitted = split(relpath(file_path, parent_path))
+
+    if splitted[0]:
+        folder_path = splitted[0].split(sep)
+
+    return "/".join([cloudinary_folder, *folder_path]).strip("/")
+

test/test_file_utils.py

@@ -0,0 +1,16 @@
+import unittest
+
+from cloudinary_cli.utils.file_utils import get_destination_folder
+
+
+class UtilsTest(unittest.TestCase):
+    def test_get_destination_folder(self):
+        """ should parse option values correctly """
+
+        self.assertEqual("1/2/3", get_destination_folder("1", "2/3/file.jpg"))
+        self.assertEqual("1", get_destination_folder("1", "file.jpg"))
+        self.assertEqual("cloudinaryfolder/myfolder/subfolder",
+                         get_destination_folder("cloudinaryfolder",
+                                                "/Users/user/myfolder/subfolder/file.jpg",
+                                                parent="/Users/user/"))
+