PILOT-2613: send lineage payload when pre-upload

app/commands/file.py

@@ -4,7 +4,6 @@
 
 import json
 import os
-import re
 
 import click
 
@@ -82,13 +81,6 @@ def cli():
     help=file_help.file_help_page(file_help.FileHELP.FILE_UPLOAD_S),
     show_default=True,
 )
-@click.option(
-    '--pipeline',
-    default=None,
-    required=False,
-    help=file_help.file_help_page(file_help.FileHELP.FILE_UPLOAD_PIPELINE),
-    show_default=True,
-)
 @click.option(
     '--zip',
     default=None,
@@ -123,7 +115,6 @@ def file_put(**kwargs):  # noqa: C901
     zone = kwargs.get('zone')
     upload_message = kwargs.get('upload_message')
     source_file = kwargs.get('source_file')
-    pipeline = kwargs.get('pipeline')
     zipping = kwargs.get('zip')
     attribute = kwargs.get('attribute')
     thread = kwargs.get('thread')
@@ -152,7 +143,6 @@ def file_put(**kwargs):  # noqa: C901
         'zone': zone,
         'upload_message': upload_message,
         'source': source_file,
-        'process_pipeline': pipeline,
         'project_code': project_code,
         'token': user.access_token,
         'attribute': attribute,
@@ -162,12 +152,6 @@ def file_put(**kwargs):  # noqa: C901
     src_file_info = validated_fieds['source_file']
     attribute = validated_fieds['attribute']
     if zone == AppConfig.Env.core_zone.lower():
-        if not pipeline:
-            # after validation, if not pipeline, provide default value
-            pipeline = AppConfig.Env.pipeline_straight_upload
-        else:
-            if not bool(re.match(r'^[a-z0-9_-]{1,20}$', pipeline)):
-                SrvErrorHandler.customized_handle(ECustomizedError.INVALID_PIPELINENAME, True)
         if not upload_message:
             upload_message = AppConfig.Env.default_upload_message
 
@@ -210,10 +194,8 @@ def file_put(**kwargs):  # noqa: C901
             'compress_zip': zipping,
             'attribute': attribute,
         }
-        if pipeline:
-            upload_event['process_pipeline'] = pipeline
         if source_file:
-            upload_event['valid_source'] = src_file_info
+            upload_event['source_id'] = src_file_info.get('id')
 
         item_ids = simple_upload(upload_event, num_of_thread=thread, output_path=output_path)
 
@@ -272,14 +254,11 @@ def validate_upload_event(event):
     zone = event.get('zone')
     upload_message = event.get('upload_message')
     source = event.get('source')
-    process_pipeline = event.get('process_pipeline')
     project_code = event.get('project_code')
     token = event.get('token')
     attribute = event.get('attribute')
     tag = event.get('tag')
-    validator = UploadEventValidator(
-        project_code, zone, upload_message, source, process_pipeline, token, attribute, tag
-    )
+    validator = UploadEventValidator(project_code, zone, upload_message, source, token, attribute, tag)
     converted_content = validator.validate_upload_event()
     return converted_content
 

app/configs/app_config.py

@@ -12,8 +12,8 @@ class Env(object):
         user_config_path = ConfigClass.config_path
         msg_path = ConfigClass.custom_path
         user_config_file = f'{user_config_path}/config.ini'
-        token_warn_need_refresh = 250  # refresh token if token is about to expire
-        token_refresh_interval = 120  # auto refresh token every 2 minutes
+        token_warn_need_refresh = 120  # refresh token if token is about to expire
+        token_refresh_interval = 90  # auto refresh token every 40 seconds
 
         # NOTE: there is a limitation on minio that
         # the multipart number is 10000. so we set

app/models/upload_form.py

@@ -2,8 +2,6 @@
 #
 # Contact Indoc Research for any questions regarding the use of this source code.
 
-from typing import List
-
 from app.services.file_manager.file_upload.models import FileObject
 
 
@@ -103,9 +101,7 @@ def generate_on_success_form(
     project_code: str,
     operator: str,
     file_object: FileObject,
-    tags: List[str],
     from_parents: str = None,
-    process_pipeline: str = None,
     upload_message: str = None,
 ):
     """
@@ -118,7 +114,6 @@ def generate_on_success_form(
         - file_object(FileObject): The object that contains the file information.
         - tags(list[str]): The tags that will be attached with file.
         - from_parents(str): indicate it is parent node.
-        - process_pipeline(str): the name of pipeline.
         - upload_message(str): the message for uploading.
     return:
         - request_payload(dict): the payload for preupload api.
@@ -135,12 +130,9 @@ def generate_on_success_form(
         'resumable_total_chunks': file_object.total_chunks,
         'resumable_total_size': file_object.total_size,
         'resumable_relative_path': file_object.parent_path,
-        'tags': tags,
     }
     if from_parents:
         request_payload['from_parents'] = from_parents
-    if process_pipeline:
-        request_payload['process_pipeline'] = process_pipeline
     if upload_message:
         request_payload['upload_message'] = upload_message
     return request_payload

app/resources/custom_error.py

@@ -66,7 +66,7 @@ class Error:
             'It means the resumable file is not the same with previous one.\n'
             'Please to double check the file content.'
         ),
-        'UNSUPPORT_TAG_MANIFEST': 'Tagging and manifest attaching are not supported for folder type.',
+        'UNSUPPORT_TAG_MANIFEST': 'Tagging, lineage and manifest attaching are not supported for folder type.',
         'INVALID_INPUT': 'Invalid input. Please try again.',
         'UNSUPPORTED_PROJECT': 'This function is not supported in the given Project %s',
         'CREATE_FOLDER_IF_NOT_EXIST': 'Target folder does not exist. Would you like to create a new folder?',

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_file = upload_event.get('valid_source')
+    source_id = upload_event.get('source_id', None)
     attribute = upload_event.get('attribute')
 
     mhandler.SrvOutPutHandler.start_uploading(input_path)
@@ -134,7 +134,7 @@ def simple_upload(  # noqa: C901
         if job_type == UploadType.AS_FILE:
             upload_file_path = [input_path.rstrip('/').lstrip() + '.zip']
             compress_folder_to_zip(input_path)
-        elif tags or attribute:
+        elif tags or attribute or source_id:
             SrvErrorHandler.customized_handle(ECustomizedError.UNSUPPORT_TAG_MANIFEST, True)
         else:
             upload_file_path = get_file_in_folder(input_path)
@@ -143,24 +143,24 @@ def simple_upload(  # noqa: C901
 
         if create_folder_flag:
             job_type = UploadType.AS_FOLDER
-            input_path = os.path.dirname(input_path)  # update the path as folder
         else:
             job_type = UploadType.AS_FILE
 
     upload_client = UploadClient(
-        input_path=input_path,
         project_code=project_code,
         zone=zone,
         job_type=job_type,
         current_folder_node=current_folder_node,
         parent_folder_id=parent_folder_id,
         regular_file=regular_file,
         tags=tags,
+        source_id=source_id,
     )
 
     # format the local path into object storage path for preupload
     file_objects = []
     target_folder = upload_event.get('target_folder', '')
+    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 + '/', '')
@@ -188,6 +188,9 @@ def simple_upload(  # noqa: C901
         # the placeholder in object storage
         pre_upload_infos.extend(upload_client.pre_upload(file_batchs, output_path))
 
+    # then output manifest file to the output path
+    upload_client.output_manifest(pre_upload_infos, output_path)
+
     # now loop over each file under the folder and start
     # the chunk upload
 
@@ -216,16 +219,13 @@ def simple_upload(  # noqa: C901
     pool.close()
     pool.join()
 
-    if source_file or attribute:
+    if attribute:
         continue_loop = True
         while continue_loop:
             # the last uploaded file
             succeed = upload_client.check_status(file_object)
             continue_loop = not succeed
             time.sleep(0.5)
-        if source_file:
-            upload_client.create_file_lineage(source_file)
-            os.remove(file_batchs[0]) if os.path.isdir(input_path) and job_type == UploadType.AS_FILE else None
 
     num_of_file = len(pre_upload_infos)
     logger.info(f'Upload Time: {time.time() - upload_start_time:.2f}s for {num_of_file:d} files')
@@ -247,7 +247,6 @@ def resume_upload(
     upload_start_time = time.time()
 
     upload_client = UploadClient(
-        input_path=manifest_json.get('file'),
         project_code=manifest_json.get('project_code'),
         zone=manifest_json.get('zone'),
         job_type='AS_FOLDER',

app/services/file_manager/file_upload/upload_client.py

@@ -28,18 +28,15 @@
 from app.services.user_authentication.token_manager import SrvTokenManager
 from app.utils.aggregated import get_file_info_by_geid
 from app.utils.aggregated import resilient_session
-from app.utils.aggregated import search_item
 
 from .exception import INVALID_CHUNK_ETAG
-from ..file_lineage import create_lineage
 
 
 class UploadClient:
     """
     Summary:
         The upload client is per upload base. it stores some immutable.
         infomation of particular upload action:
-         - input_path: the path that user inputs. can be a folder or file.
          - project_code: the unique code of project.
          - zone: data zone. can be greenroom or core.
          - upload_message:
@@ -49,20 +46,18 @@ class UploadClient:
 
     def __init__(
         self,
-        input_path: str,
         project_code: str,
         parent_folder_id: str,
         zone: str = AppConfig.Env.green_zone,
         upload_message: str = 'cli straight upload',
         job_type: str = UploadType.AS_FILE,
-        process_pipeline: str = None,
         current_folder_node: str = '',
         regular_file: str = True,
         tags: list = None,
+        source_id: str = '',
     ):
         self.user = UserConfig()
         self.operator = self.user.username
-        self.input_path = input_path
         self.upload_message = upload_message
         self.chunk_size = AppConfig.Env.chunk_size  # remove
         self.base_url = {
@@ -79,11 +74,12 @@ def __init__(
         self.zone = zone
         self.job_type = job_type
         self.project_code = project_code
-        self.process_pipeline = process_pipeline
         self.current_folder_node = current_folder_node
         self.parent_folder_id = parent_folder_id
         self.regular_file = regular_file
+        # tags and souce_id are only allowed in file uplaod
         self.tags = tags
+        self.source_id = source_id
 
         # the flag to indicate if all upload process finished
         # then the token refresh loop will end
@@ -94,7 +90,7 @@ def generate_meta(self, local_path: str) -> Tuple[int, int]:
         Summary:
             The function is to generate chunk upload meatedata for a file.
         Parameter:
-            - input_path: The path of the local file eg. a/b/c.txt.
+            - local_path: The path of the local file eg. a/b/c.txt.
         return:
             - total_size: the size of file.
             - total_chunks: the number of chunks will be uploaded.
@@ -176,13 +172,13 @@ def pre_upload(self, file_objects: List[FileObject], output_path: str) -> List[F
             'current_folder_node': self.current_folder_node,
             'parent_folder_id': self.parent_folder_id,
             'folder_tags': self.tags,
+            'source_id': self.source_id,
             '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}
@@ -196,9 +192,6 @@ def pre_upload(self, file_objects: List[FileObject], output_path: str) -> List[F
                 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)
-
             mhandler.SrvOutPutHandler.preupload_success()
             return file_objets
         elif response.status_code == 403:
@@ -372,9 +365,7 @@ def on_succeed(self, file_object: FileObject, tags: List[str], chunk_result: Lis
                 self.project_code,
                 self.operator,
                 file_object,
-                tags,
                 [],
-                process_pipeline=self.process_pipeline,
                 upload_message=self.upload_message,
             )
             headers = {
@@ -397,36 +388,6 @@ def on_succeed(self, file_object: FileObject, tags: List[str], chunk_result: Lis
 
             time.sleep(AppConfig.Env.resilient_retry_interval * (i + 1))
 
-    @require_valid_token()
-    def create_file_lineage(self, source_file: dict, new_file_object: FileObject):
-        """
-        Summary:
-            The function is to create a lineage with source file.
-        Parameter:
-            - source_file(str): the file object that indicate the exist data to link with.
-            - new_file_object(FileObject): the new object just uploaded.
-        return:
-            - bool: if job success or not.
-        """
-
-        if source_file and self.zone == AppConfig.Env.core_zone:
-            child_rel_path = new_file_object.object_path
-            child_item = search_item(self.project_code, self.zone, child_rel_path, 'file')
-            child_file = child_item['result']
-            parent_file_geid = source_file['id']
-            child_file_geid = child_file['id']
-            lineage_event = {
-                'input_id': parent_file_geid,
-                'output_id': child_file_geid,
-                'input_path': os.path.join(source_file['parent_path'], source_file['name']),
-                'output_path': os.path.join(child_file['parent_path'], child_file['name']),
-                'project_code': self.project_code,
-                'action_type': self.process_pipeline,
-                'operator': self.operator,
-                'token': self.user.access_token,
-            }
-            create_lineage(lineage_event)
-
     def check_status(self, file_object: FileObject) -> bool:
         """
         Summary: