diff --git a/app/commands/file.py b/app/commands/file.py index c17ed119..955cb174 100644 --- a/app/commands/file.py +++ b/app/commands/file.py @@ -171,11 +171,24 @@ 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, @@ -183,9 +196,11 @@ def file_put(**kwargs): # noqa: C901 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, diff --git a/app/models/upload_form.py b/app/models/upload_form.py index 9a594229..d51e7713 100644 --- a/app/models/upload_form.py +++ b/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 is the folder user key in - # eg. a/b/ . the is files under eg a/b/c/d.txt. The - # path in object storage will be /c/d.txt - # - if use input as a file then is the file user key in eg. - # a/b/c/d.txt. the will be same as it. The path in - # object storage will be /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 : . 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, diff --git a/app/services/file_manager/file_upload/file_upload.py b/app/services/file_manager/file_upload/file_upload.py index 41bb6718..aded3707 100644 --- a/app/services/file_manager/file_upload/file_upload.py +++ b/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,13 +111,13 @@ 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) @@ -123,52 +125,56 @@ def simple_upload( # noqa: C901 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'), - [], ) ) diff --git a/app/services/file_manager/file_upload/models.py b/app/services/file_manager/file_upload/models.py index 6f548652..e722d8e7 100644 --- a/app/services/file_manager/file_upload/models.py +++ b/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]: """ diff --git a/app/services/file_manager/file_upload/upload_client.py b/app/services/file_manager/file_upload/upload_client.py index 094ff5c0..cbeaf7f1 100644 --- a/app/services/file_manager/file_upload/upload_client.py +++ b/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) diff --git a/tests/app/services/file_manager/file_upload/test_file_upload.py b/tests/app/services/file_manager/file_upload/test_file_upload.py index 2faca549..5044f9c4 100644 --- a/tests/app/services/file_manager/file_upload/test_file_upload.py +++ b/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', diff --git a/tests/app/services/file_manager/file_upload/test_model.py b/tests/app/services/file_manager/file_upload/test_model.py index da612bf7..7b9924ab 100644 --- a/tests/app/services/file_manager/file_upload/test_model.py +++ b/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 diff --git a/tests/app/services/file_manager/file_upload/test_upload_client.py b/tests/app/services/file_manager/file_upload/test_upload_client.py index 1db70077..03ce331a 100644 --- a/tests/app/services/file_manager/file_upload/test_upload_client.py +++ b/tests/app/services/file_manager/file_upload/test_upload_client.py @@ -38,7 +38,7 @@ def test_check_status_success(httpx_mock, mocker): status_code=200, ) - test_obj = FileObject('test', 'test', 'test', 'test', 'test', []) + test_obj = FileObject('test', 'test', 'test', 'test', 'test') result = upload_client.check_status(test_obj) assert result is True @@ -57,7 +57,7 @@ def test_check_status_fail(httpx_mock, mocker): status_code=200, ) - test_obj = FileObject('test', 'test', 'test', 'test', 'test', []) + test_obj = FileObject('test', 'test', 'test', 'test', 'test') result = upload_client.check_status(test_obj) assert result is False @@ -72,7 +72,7 @@ def test_chunk_upload(httpx_mock, mocker): httpx_mock.add_response(method='PUT', url=test_presigned_url, json={'result': ''}) mocker.patch('app.services.file_manager.file_upload.models.FileObject.generate_meta', return_value=(1, 1)) - test_obj = FileObject('test', 'test', 'test', 'test', 'test', []) + test_obj = FileObject('test', 'test', 'test', 'test', 'test') res = upload_client.upload_chunk(test_obj, 0, b'1') assert test_obj.progress_bar.n == 1 @@ -113,7 +113,7 @@ def test_resumable_pre_upload_success(httpx_mock, mocker): upload_client = UploadClient('test', 'project_code', 'parent_folder_id') 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') url = AppConfig.Connections.url_bff + f'/v1/project/{upload_client.project_code}/files/resumable' httpx_mock.add_response( @@ -135,7 +135,7 @@ def test_resumable_pre_upload_failed_with_404(httpx_mock, mocker): upload_client = UploadClient('test', 'project_code', 'parent_folder_id') 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') url = AppConfig.Connections.url_bff + f'/v1/project/{upload_client.project_code}/files/resumable' httpx_mock.add_response( @@ -157,7 +157,7 @@ def test_output_manifest_success(mocker): upload_client = UploadClient('test', 'project_code', 'parent_folder_id') json_dump_mocker = mocker.patch('json.dump', return_value=None) 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') res = upload_client.output_manifest([test_obj], 'test')