merge back/staging release 18042023

app/commands/file.py

@@ -195,7 +195,7 @@ def file_put(**kwargs):  # noqa: C901
             'attribute': attribute,
         }
         if source_file:
-            upload_event['source_id'] = src_file_info.get('id')
+            upload_event['source_id'] = src_file_info.get('id', '')
 
         item_ids = simple_upload(upload_event, num_of_thread=thread, output_path=output_path)
 

app/resources/custom_help.py

@@ -6,9 +6,11 @@
 class HelpPage:
     page = {
         'update': {
-            'version': '2.2.0',
-            '1': 'CLI supports to perform multi-threading upload for file/folders',
-            '2': 'CLI supports to perform resumable upload for single file',
+            'version': '2.3.0',
+            '1': 'The logic of normal upload and resumble are splited. '
+            'add new command for resumable upload as `pilotcli file resume -r manifest.json`',
+            '2': 'The manifest file will be output for both file/folder upload',
+            '3': 'Optimize logic, input and error message',
         },
         'dataset': {
             'DATASET_DOWNLOAD': 'Download a dataset or a particular version of a dataset.',

app/services/file_manager/file_upload/file_upload.py

@@ -123,7 +123,7 @@ def simple_upload(  # noqa: C901
     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_id = upload_event.get('source_id', None)
+    source_id = upload_event.get('source_id', '')
     attribute = upload_event.get('attribute')
 
     mhandler.SrvOutPutHandler.start_uploading(input_path)
@@ -163,7 +163,7 @@ def simple_upload(  # noqa: C901
     input_path = os.path.dirname(input_path)
     for file in upload_file_path:
         # first remove the input path from the file path
-        file_path_sub = file.replace(input_path + '/', '')
+        file_path_sub = file.replace(input_path + '/', '') if input_path else file
         object_path = os.path.join(target_folder, file_path_sub)
 
         # generate a placeholder for each file
@@ -256,28 +256,44 @@ def resume_upload(
     )
 
     # check files in manifest if some of them are already uploaded
-    item_ids = []
+    unfinished_items = []
     all_files = manifest_json.get('file_objects')
+    item_ids = []
     for item_id in all_files:
         item_ids.append(item_id)
-    items = get_file_info_by_geid(item_ids)
 
-    unfinished_items = []
-    for x in items:
-        if x.get('result').get('status') == ItemStatus.REGISTERED:
-            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'),
+    # 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(all_files) / AppConfig.Env.upload_batch_size)
+    # here is list of pre upload result. We decided to call pre upload api by batch
+    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 = item_ids[start_index:end_index]
+        items = get_file_info_by_geid(file_batchs)
+
+        # get the detail of item to see if the file is already uploaded
+        unfinished_files = []
+        for x in items:
+            if x.get('result').get('status') == ItemStatus.REGISTERED:
+                file_info = all_files.get(x.get('result').get('id'))
+                unfinished_files.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'),
+                    )
                 )
-            )
 
-    # then for the rest of the files, check if any chunks are already uploaded
-    unfinished_items = upload_client.resume_upload(unfinished_items)
+        # then for the rest of the files, check if any chunks are already uploaded
+        mhandler.SrvOutPutHandler.resume_check_in_progress()
+        if len(unfinished_files) > 0:
+            unfinished_items.extend(upload_client.resume_upload(unfinished_files))
+
+    mhandler.SrvOutPutHandler.resume_warning(len(unfinished_items))
+    mhandler.SrvOutPutHandler.resume_check_success()
 
     # lastly, start resumable upload for the rest of the chunks
     # thread number +1 reserve one thread to refresh token

app/services/file_manager/file_upload/upload_client.py

@@ -114,7 +114,6 @@ def resume_upload(self, unfinished_file_objects: List[FileObject]) -> List[FileO
                 - local_path(str): the local path of file.
                 - chunk_info(dict): the mapping for chunks that already been uploaded.
         """
-        mhandler.SrvOutPutHandler.resume_warning(len(unfinished_file_objects))
 
         headers = {'Authorization': 'Bearer ' + self.user.access_token, 'Session-ID': self.user.session_id}
         url = AppConfig.Connections.url_bff + f'/v1/project/{self.project_code}/files/resumable'
@@ -142,7 +141,6 @@ def resume_upload(self, unfinished_file_objects: List[FileObject]) -> List[FileO
             file_obj = rid_file_object_map.get(uploaded_info.get('resumable_id'))
             # update the chunk info
             file_obj.uploaded_chunks = uploaded_info.get('chunks_info')
-        mhandler.SrvOutPutHandler.resume_check_success()
 
         return unfinished_file_objects
 

app/services/output_manager/message_handler.py

@@ -133,6 +133,11 @@ def resume_check_success():
         """e.g. notify the resumable check succeed."""
         return logger.info('Resumable upload check complete.')
 
+    @staticmethod
+    def resume_check_in_progress():
+        """e.g. notify the resumable check succeed."""
+        return logger.info('Resumable upload check in progress.')
+
     @staticmethod
     def resume_warning(num_of_files: int):
         """e.g. notify the user if they comfirm the resumable upload."""

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "app"
-version = "2.2.4"
+version = "2.4.0a0"
 description = "This service is designed to support pilot platform"
 authors = ["Indoc Research"]