Pilot 2392: refactor the upload logic

app/commands/file.py

@@ -171,21 +171,36 @@ def file_put(**kwargs):  # noqa: C901
         if not upload_message:
             upload_message = AppConfig.Env.default_upload_message
 
+    # for the path formating there will be following cases:
+    # - file:
+    #   1. the project path exist, then will be AS_FILE. nothing will be changed.
+    #      current_folder_node will be empty string.
+    #   2. the project path not exist, then will be AS_FOLDER. the current_folder_node will
+    #      be the parent folder node + the shortest non-exist folder. (like one level down).
+    # - folder:
+    #   1. the project path exist, then will be AS_FOLDER. the current folder node will be
+    #      the one that user input.
+    #   2. the project path not exist, then will be AS_FOLDER. the current folder node will
+    #      be the parent folder node + the shortest non-exist folder. (like one level down).
+
     # Unique Paths
     paths = set(paths)
     # the loop will read all input path(folder or files)
     # and process them one by one
     for f in paths:
+        # so this function will always return the furthest folder node as current_folder_node+parent_folder_id
         current_folder_node, parent_folder, create_folder_flag, result_file = assemble_path(
             f,
             target_folder,
             project_code,
             zone,
             zipping,
         )
+
         upload_event = {
             'project_code': project_code,
-            'file': f,
+            'target_folder': target_folder,
+            'file': f.rstrip('/'),  # remove the ending slash
             'tags': tag if tag else [],
             'zone': zone,
             'upload_message': upload_message,

app/models/upload_form.py

@@ -2,13 +2,9 @@
 #
 # Contact Indoc Research for any questions regarding the use of this source code.
 
-from os.path import basename
-from os.path import dirname
-from os.path import join
 from typing import List
 
 from app.services.file_manager.file_upload.models import FileObject
-from app.services.file_manager.file_upload.models import UploadType
 
 
 class FileUploadForm:
@@ -103,67 +99,6 @@ def metadatas(self, metadatas):
         self._attribute_map['metadatas'] = metadatas
 
 
-def generate_pre_upload_form(
-    project_code: str,
-    operator: str,
-    local_file_paths: List[str],
-    input_path: str,
-    zone: str,
-    job_type: UploadType,
-    current_folder: str = '',
-) -> tuple[dict, dict]:
-    """
-    Summary:
-        The function is to generate the preupload payload for api. The operation
-        is per batch that it will try to generate one payload for all files.
-    Parameter:
-        - project_code(str): The unique identifier for project.
-        - operator(str): The name of operator.
-        - local_file_paths(list[str]): The list of name for input files.
-        - input_path: The path specified by user, if it is folder, it will be like
-            a/b . If it is a file it will be same as local_file_paths eg. a/b/c.txt.
-        - zone(str): The zone of user try to upload to.
-        - job_type(UploadType): the upload type, AS_FOLDER or AS_FILE.
-        - current_folder(str): the folder path on object storage that user specified.
-    return:
-        - request_payload(dict): the payload for preupload api.
-        - local_file_mapping(dict): the mapping from object path into local path.
-    """
-    data, local_file_mapping = [], {}
-    for file_local_path in local_file_paths:
-        # the rule here is:
-        # - if use input as a folder then <input_path> is the folder user key in
-        #   eg. a/b/ . the <local_file_paths> is files under eg a/b/c/d.txt. The
-        #   path in object storage will be <current_folder>/c/d.txt
-        # - if use input as a file then <input_path> is the file user key in eg.
-        #   a/b/c/d.txt. the <local_file_paths> will be same as it. The path in
-        #   object storage will be <current_folder>/d.txt
-        if job_type == UploadType.AS_FOLDER:
-            file_relative_path = file_local_path.replace(input_path + '/', '')
-            object_path = join(current_folder, file_relative_path)
-            parent_path, file_name = dirname(object_path), basename(object_path)
-        else:
-            file_name = basename(file_local_path)
-            parent_path = current_folder
-
-        data.append({'resumable_filename': file_name, 'resumable_relative_path': parent_path})
-        # make a mapping as <object_path>: <local_path>. This will be returned
-        # and used in chunk upload api.
-        object_path = join(parent_path, file_name)
-        local_file_mapping.update({object_path: file_local_path})
-
-    request_payload = {
-        'project_code': project_code,
-        'operator': operator,
-        'job_type': str(job_type),
-        'zone': zone,
-        'current_folder_node': current_folder,
-        'data': data,
-    }
-
-    return request_payload, local_file_mapping
-
-
 def generate_on_success_form(
     project_code: str,
     operator: str,

app/services/file_manager/file_upload/file_upload.py

@@ -69,13 +69,18 @@ def assemble_path(
 
     current_file_path = target_folder + '/' + f.rstrip('/').split('/')[-1]
     result_file = current_file_path
+    if zipping:
+        result_file = result_file + '.zip'
 
     # set name folder as first parent folder
-    name_folder = current_file_path.split('/')[0]
+    name_folder = target_folder.split('/')[0]
     parent_folder = search_item(project_code, zone, name_folder, 'name_folder')
     parent_folder = parent_folder.get('result')
-    create_folder_flag = False
 
+    # if f input is a file then current_folder_node is target_folder
+    # otherwise it is target_folder + f input name
+    current_folder_node = target_folder if os.path.isfile(f) else current_file_path
+    create_folder_flag = False
     if len(current_file_path.split('/')) > 2:
         sub_path = target_folder.split('/')
         for index in range(len(sub_path) - 1):
@@ -85,7 +90,7 @@ def assemble_path(
             # find the longest existing folder as parent folder
             # if user input a path that need to create some folders
             if not res.get('result'):
-                current_file_path = folder_path
+                current_folder_node = folder_path
                 click.confirm(customized_error_msg(ECustomizedError.CREATE_FOLDER_IF_NOT_EXIST), abort=True)
                 create_folder_flag = True
                 break
@@ -97,10 +102,7 @@ def assemble_path(
     if not parent_folder:
         SrvErrorHandler.customized_handle(ECustomizedError.PERMISSION_DENIED, True)
 
-    if zipping:
-        result_file = result_file + '.zip'
-
-    return current_file_path, parent_folder, create_folder_flag, result_file
+    return current_folder_node, parent_folder, create_folder_flag, result_file
 
 
 def simple_upload(  # noqa: C901
@@ -109,66 +111,70 @@ def simple_upload(  # noqa: C901
     output_path: str = None,
 ):
     upload_start_time = time.time()
-    my_file = upload_event.get('file')
+    input_path = upload_event.get('file')
     project_code = upload_event.get('project_code')
     tags = upload_event.get('tags')
     zone = upload_event.get('zone')
     # process_pipeline = upload_event.get('process_pipeline', None)
     # upload_message = upload_event.get('upload_message')
-    target_folder = upload_event.get('current_folder_node', '')
+    current_folder_node = upload_event.get('current_folder_node', '')
     parent_folder_id = upload_event.get('parent_folder_id', '')
     create_folder_flag = upload_event.get('create_folder_flag', False)
     compress_zip = upload_event.get('compress_zip', False)
     regular_file = upload_event.get('regular_file', True)
     source_file = upload_event.get('valid_source')
     attribute = upload_event.get('attribute')
 
-    mhandler.SrvOutPutHandler.start_uploading(my_file)
-    # TODO: PILOT-2392 simplify the logic under
+    mhandler.SrvOutPutHandler.start_uploading(input_path)
     # if the input request zip folder then process the path as single file
     # otherwise read throught the folder to get path underneath
-    if os.path.isdir(my_file):
+    if os.path.isdir(input_path):
         job_type = UploadType.AS_FILE if compress_zip else UploadType.AS_FOLDER
         if job_type == UploadType.AS_FILE:
-            upload_file_path = [my_file.rstrip('/').lstrip() + '.zip']
-            target_folder = '/'.join(target_folder.split('/')[:-1]).rstrip('/')
-            compress_folder_to_zip(my_file)
+            upload_file_path = [input_path.rstrip('/').lstrip() + '.zip']
+            compress_folder_to_zip(input_path)
         elif tags or attribute:
             SrvErrorHandler.customized_handle(ECustomizedError.UNSUPPORT_TAG_MANIFEST, True)
         else:
-            upload_file_path = get_file_in_folder(my_file)
+            upload_file_path = get_file_in_folder(input_path)
     else:
-        upload_file_path = [my_file]
+        upload_file_path = [input_path]
 
         if create_folder_flag:
             job_type = UploadType.AS_FOLDER
-            my_file = os.path.dirname(my_file)  # update the path as folder
+            input_path = os.path.dirname(input_path)  # update the path as folder
         else:
-            target_folder = '/'.join(target_folder.split('/')[:-1]).rstrip('/')
             job_type = UploadType.AS_FILE
 
     upload_client = UploadClient(
-        input_path=my_file,
+        input_path=input_path,
         project_code=project_code,
         zone=zone,
         job_type=job_type,
-        current_folder_node=target_folder,
+        current_folder_node=current_folder_node,
         parent_folder_id=parent_folder_id,
         regular_file=regular_file,
         tags=tags,
     )
 
+    # format the local path into object storage path for preupload
+    file_objects = []
+    target_folder = upload_event.get('target_folder', '')
+    for file in upload_file_path:
+        # first remove the input path from the file path
+        file_path_sub = file.replace(input_path + '/', '')
+        object_path = os.path.join(target_folder, file_path_sub)
+        file_objects.append(FileObject(object_path, file))
+
     # here add the batch of 500 per loop, the pre upload api cannot
     # process very large amount of file at same time. otherwise it will timeout
-    num_of_batchs = math.ceil(len(upload_file_path) / AppConfig.Env.upload_batch_size)
+    num_of_batchs = math.ceil(len(file_objects) / AppConfig.Env.upload_batch_size)
     # here is list of pre upload result. We decided to call pre upload api by batch
-    # the result will store as (UploaderObject, preupload_id_mapping)
     pre_upload_infos = []
-
     for batch in range(0, num_of_batchs):
         start_index = batch * AppConfig.Env.upload_batch_size
         end_index = (batch + 1) * AppConfig.Env.upload_batch_size
-        file_batchs = upload_file_path[start_index:end_index]
+        file_batchs = file_objects[start_index:end_index]
 
         # sending the pre upload request to generate
         # the placeholder in object storage
@@ -212,7 +218,7 @@ def simple_upload(  # noqa: C901
             time.sleep(0.5)
         if source_file:
             upload_client.create_file_lineage(source_file)
-            os.remove(file_batchs[0]) if os.path.isdir(my_file) and job_type == UploadType.AS_FILE else None
+            os.remove(file_batchs[0]) if os.path.isdir(input_path) and job_type == UploadType.AS_FILE else None
 
     num_of_file = len(upload_file_path)
     logger.info(f'Upload Time: {time.time() - upload_start_time:.2f}s for {num_of_file:d} files')
@@ -254,12 +260,11 @@ def resume_upload(
             file_info = all_files.get(x.get('result').get('id'))
             unfinished_items.append(
                 FileObject(
+                    file_info.get('object_path'),
+                    file_info.get('local_path'),
                     file_info.get('resumable_id'),
                     file_info.get('job_id'),
                     file_info.get('item_id'),
-                    file_info.get('object_path'),
-                    file_info.get('local_path'),
-                    [],
                 )
             )
 

app/services/file_manager/file_upload/models.py

@@ -66,7 +66,12 @@ class FileObject:
     progress_bar = None
 
     def __init__(
-        self, resumable_id: str, job_id: str, item_id: str, object_path: str, local_path: str, uploaded_chunks: List
+        self,
+        object_path: str,
+        local_path: str,
+        resumable_id: str = None,
+        job_id: str = None,
+        item_id: str = None,
     ) -> None:
         # object storage info
         self.resumable_id = resumable_id
@@ -80,7 +85,7 @@ def __init__(
         self.total_size, self.total_chunks = self.generate_meta(local_path)
 
         # resumable info
-        self.uploaded_chunks = uploaded_chunks
+        self.uploaded_chunks = {}
 
     def generate_meta(self, local_path: str) -> Tuple[int, int]:
         """

app/services/file_manager/file_upload/upload_client.py

@@ -151,7 +151,7 @@ def resume_upload(self, unfinished_file_objects: List[FileObject]) -> List[FileO
         return unfinished_file_objects
 
     @require_valid_token()
-    def pre_upload(self, local_file_paths: List[str], output_path: str) -> List[FileObject]:
+    def pre_upload(self, file_objects: List[FileObject], output_path: str) -> List[FileObject]:
         """
         Summary:
             The function is to initiate all the multipart upload.
@@ -168,32 +168,33 @@ def pre_upload(self, local_file_paths: List[str], output_path: str) -> List[File
 
         headers = {'Authorization': 'Bearer ' + self.user.access_token, 'Session-ID': self.user.session_id}
         url = AppConfig.Connections.url_bff + '/v1/project/{}/files'.format(self.project_code)
-        # the file mapping is a dictionary that present the map from object storage path
-        # with local file path. It will be used in chunk upload api.
-        payload, file_mapping = uf.generate_pre_upload_form(
-            self.project_code,
-            self.operator,
-            local_file_paths,
-            self.input_path,
-            zone=self.zone,
-            job_type=self.job_type,
-            current_folder=self.current_folder_node,
-        )
-
-        payload.update({'parent_folder_id': self.parent_folder_id})
-        payload.update({'folder_tags': self.tags})
+        payload = {
+            'project_code': self.project_code,
+            'operator': self.operator,
+            'job_type': str(self.job_type),
+            'zone': self.zone,
+            'current_folder_node': self.current_folder_node,
+            'parent_folder_id': self.parent_folder_id,
+            'folder_tags': self.tags,
+            'data': [
+                {'resumable_filename': x.file_name, 'resumable_relative_path': x.parent_path} for x in file_objects
+            ],
+        }
+
         response = resilient_session().post(url, json=payload, headers=headers, timeout=None)
+
         if response.status_code == 200:
             result = response.json().get('result')
+            file_mapping = {x.object_path: x for x in file_objects}
             file_objets = []
             for job in result:
                 object_path = job.get('target_names')[0]
-                resumable_id = job.get('payload').get('resumable_identifier')
-                item_id = job.get('payload').get('item_id')
-                job_id = job.get('job_id')
-                file_objets.append(
-                    FileObject(resumable_id, job_id, item_id, object_path, file_mapping.get(object_path), {})
-                )
+                # get the file object from mapping and update the attribute
+                file_object = file_mapping.get(object_path)
+                file_object.resumable_id = job.get('payload').get('resumable_identifier')
+                file_object.item_id = job.get('payload').get('item_id')
+                file_object.job_id = job.get('job_id')
+                file_objets.append(file_object)
 
             # then output manifest file to the output path
             self.output_manifest(file_objets, output_path)

tests/app/services/file_manager/file_upload/test_file_upload.py

@@ -161,7 +161,7 @@ def test_dont_allow_attribute_attaching_when_folder_upload(mocker, capfd):
 
 def test_resume_upload(mocker):
     mocker.patch('app.services.file_manager.file_upload.models.FileObject.generate_meta', return_value=(1, 1))
-    test_obj = FileObject('resumable_id', 'job_id', 'item_id', 'object/path', 'local_path', [])
+    test_obj = FileObject('object/path', 'local_path', 'resumable_id', 'job_id', 'item_id')
 
     manifest_json = {
         'project_code': 'project_code',

tests/app/services/file_manager/file_upload/test_model.py

@@ -9,7 +9,7 @@
 def test_file_upload_model_update_progress_bar(mocker):
     mocker.patch('app.services.file_manager.file_upload.models.getsize', return_value=100)
 
-    file_obj = FileObject('test', 'test', 'test', 'test', 'test', [])
+    file_obj = FileObject('test', 'test', 'test', 'test', 'test')
     file_obj.update_progress(1)
 
     assert file_obj.progress_bar is not None
@@ -19,7 +19,7 @@ def test_file_upload_model_update_progress_bar(mocker):
 def test_file_upload_model_close_progress_bar(mocker):
     mocker.patch('app.services.file_manager.file_upload.models.getsize', return_value=100)
 
-    file_obj = FileObject('test', 'test', 'test', 'test', 'test', [])
+    file_obj = FileObject('test', 'test', 'test', 'test', 'test')
     file_obj.close_progress()
 
     assert file_obj.progress_bar is None
@@ -29,7 +29,7 @@ def test_file_upload_model_generate_meta(mocker):
     AppConfig.Env.chunk_size = 10
     mocker.patch('app.services.file_manager.file_upload.models.getsize', return_value=100)
 
-    file_obj = FileObject('test', 'test', 'test', 'test', 'test', [])
+    file_obj = FileObject('test', 'test', 'test', 'test', 'test')
     total_size, total_chunks = file_obj.generate_meta('test')
 
     assert total_size == 100