Pilot 4090: add new command to download file metadata to local

app/commands/entry_point.py

@@ -20,6 +20,7 @@
 from .file import file_download
 from .file import file_export_manifest
 from .file import file_list
+from .file import file_metadata_download
 from .file import file_put
 from .file import file_resume
 
@@ -72,6 +73,7 @@ def user_group():
 file_group.add_command(file_list)
 file_group.add_command(file_download)
 file_group.add_command(file_resume)
+file_group.add_command(file_metadata_download)
 project_group.add_command(project_list_all)
 user_group.add_command(login)
 user_group.add_command(logout)

app/commands/file.py

@@ -16,6 +16,7 @@
 from app.services.file_manager.file_download.download_client import SrvFileDownload
 from app.services.file_manager.file_list import SrvFileList
 from app.services.file_manager.file_manifests import SrvFileManifests
+from app.services.file_manager.file_metadata.file_metadata_client import FileMetaClient
 from app.services.file_manager.file_upload.file_upload import assemble_path
 from app.services.file_manager.file_upload.file_upload import resume_upload
 from app.services.file_manager.file_upload.file_upload import simple_upload
@@ -435,3 +436,60 @@ def file_download(**kwargs):
         for item in item_res:
             srv_download = SrvFileDownload(zone, interactive)
             srv_download.simple_download_file(output_path, [item])
+
+
+@click.command(name='metadata')
+@click.argument('file_path', type=click.STRING)
+@click.option(
+    '-z',
+    '--zone',
+    default=AppConfig.Env.green_zone,
+    required=True,
+    help=file_help.file_help_page(file_help.FileHELP.FILE_META_Z),
+    show_default=False,
+)
+@click.option(
+    '-g',
+    '--general',
+    default=None,
+    required=True,
+    help=file_help.file_help_page(file_help.FileHELP.FILE_META_G),
+    show_default=True,
+)
+@click.option(
+    '-a',
+    '--attribute',
+    default=None,
+    required=True,
+    help=file_help.file_help_page(file_help.FileHELP.FILE_META_A),
+    show_default=True,
+)
+@click.option(
+    '-t',
+    '--tag',
+    default=None,
+    required=True,
+    help=file_help.file_help_page(file_help.FileHELP.FILE_META_T),
+    show_default=True,
+)
+@require_valid_token()
+@doc(file_help.file_help_page(file_help.FileHELP.FILE_META))
+def file_metadata_download(**kwargs):
+    '''
+    Summary:
+        Download metadata of a file including general, attribute and tag.
+    '''
+
+    file_path = kwargs.get('file_path')
+    zone = kwargs.get('zone')
+    general_folder = kwargs.get('general').rstrip('/')
+    attribute_folder = kwargs.get('attribute').rstrip('/')
+    tag_folder = kwargs.get('tag').rstrip('/')
+
+    # user = UserConfig()
+    # Check zone and upload-message
+    zone = get_zone(zone) if zone else AppConfig.Env.green_zone.lower()
+    file_meta_client = FileMetaClient(zone, file_path, general_folder, attribute_folder, tag_folder)
+    file_meta_client.download_file_metadata()
+
+    message_handler.SrvOutPutHandler.metadata_download_success()

app/configs/app_config.py

@@ -45,3 +45,4 @@ class Connections:
         url_keycloak_token = f'{ConfigClass.url_keycloak}/token'
         url_bff = ConfigClass.url_bff
         url_base = ConfigClass.base_url
+        url_portal = ConfigClass.url_portal

app/configs/config.py

@@ -41,6 +41,10 @@ def base_url(self) -> str:
     def url_bff(self) -> str:
         return f'{self.base_url}/cli'
 
+    @computed_field
+    def url_portal(self) -> str:
+        return f'{self.base_url}/portal'
+
     @computed_field
     def url_keycloak_realm(self) -> str:
         return f'https://iam.{self.domain}/realms/pilot'

app/resources/custom_error.py

@@ -93,6 +93,8 @@ class Error:
         'INVALID_FOLDER': 'Provided folder does not exist',
         'INVALID_NAMEFOLDER': 'User name folder is missing or provided user name folder does not exist',
         'INVALID_DOWNLOAD': 'Invalid download, file/folder not exist or folder is empty: %s',
+        # file metadata related error
+        'LOCAL_METADATA_FILE_EXISTS': 'Following metadata file already exists in the local directory: ',
         'VERSION_NOT_EXIST': 'Version not available: %s',
         'DATASET_NOT_EXIST': 'Dataset not found in your dataset list',
         'DATASET_PERMISSION': 'You do not have permission to access this dataset',

app/resources/custom_help.py

@@ -54,6 +54,11 @@ class HelpPage:
                 "The processed pipeline of your processed files. [only used with '--source' option]"
             ),
             'FILE_UPLOAD_ZIP': 'Upload folder as a compressed zip file.',
+            'FILE_META': 'Download metadata file of a given file in target zone.',
+            'FILE_META_Z': 'Target Zone (i.e., core/greenroom)',
+            'FILE_META_G': 'The location of general metadata file',
+            'FILE_META_A': 'The location of attribute metadata file',
+            'FILE_META_T': 'The location of tag metadata file',
         },
         'config': {
             'SET_CONFIG': 'Chose config file and set for cli.',

app/services/file_manager/file_metadata/__init__.py

@@ -0,0 +1,3 @@
+# Copyright (C) 2022-2023 Indoc Systems
+#
+# Contact Indoc Systems for any questions regarding the use of this source code.

app/services/file_manager/file_metadata/file_metadata_client.py

@@ -0,0 +1,125 @@
+# Copyright (C) 2022-2023 Indoc Systems
+#
+# Contact Indoc Systems for any questions regarding the use of this source code.
+
+import json
+from os import makedirs
+from os.path import basename
+from os.path import dirname
+from os.path import exists
+from os.path import join
+from sys import exit
+from typing import Any
+from typing import Dict
+from typing import List
+from typing import Union
+
+import click
+from click.exceptions import Abort
+
+import app.services.output_manager.message_handler as message_handler
+from app.services.output_manager.error_handler import ECustomizedError
+from app.services.output_manager.error_handler import customized_error_msg
+from app.utils.aggregated import get_attribute_template_by_id
+from app.utils.aggregated import search_item
+
+
+class FileMetaClient:
+    """
+    Summary:
+        A client for interacting with file metadata. currently support to download
+        file metadata from metadata service.
+    """
+
+    def __init__(
+        self,
+        zone: str,
+        file_path: str,
+        general_folder: str,
+        attribute_folder: str,
+        tag_folder: str,
+    ) -> None:
+        """
+        Summary:
+            Initialize file metadata client.
+        Parameters:
+            zone (str): zone.
+            file_path (str): file path.
+            general_folder (str): local folder of general metadata eg. item_id, project_code.
+            attribute_folder (str): local folder of attribute metadata.
+            tag_folder (str): local folder of tag metadata.
+        """
+
+        self.zone = zone
+        self.file_path = file_path
+        self.project_code, self.object_path = self.file_path.split('/', 1)
+        # only get the name regardless of the extension
+        self.file_name = basename(self.object_path).rsplit('.', 1)[0]
+
+        # location of metadata files
+        self.general_location = join(general_folder, f'{self.file_name}-general.json')
+        self.attribute_location = join(attribute_folder, f'{self.file_name}-attribute.json')
+        self.tag_location = join(tag_folder, f'{self.file_name}-tag.json')
+        self._check_duplication(self.general_location, self.attribute_location, self.tag_location)
+
+    def _check_duplication(self, general_loc: str, attribute_loc: str, tag_loc: str) -> None:
+        """
+        Summary:
+            Check if the metadata files already exist in location system
+            and ask user whether to overwrite.
+        """
+
+        file_dict = {'general': general_loc, 'attribute': attribute_loc, 'tag': tag_loc}
+
+        # check if the manifest file exists and ask user whether to overwrite
+        try:
+            duplicate_error = customized_error_msg(ECustomizedError.LOCAL_METADATA_FILE_EXISTS)
+            overwrite_check = False
+            for metadata_name, location in file_dict.items():
+                if exists(location):
+                    overwrite_check = True
+                    duplicate_error = duplicate_error + f'\n - {metadata_name}: {location}'
+
+            if overwrite_check:
+                duplicate_error = duplicate_error + '\nDo you want to overwrite the existing file?'
+                click.confirm(duplicate_error, abort=True)
+        except Abort:
+            message_handler.SrvOutPutHandler.cancel_metadata_download()
+            exit(1)
+
+    def save_file_metadata(self, file_loc: str, metadata: Union[dict, list]) -> None:
+        """
+        Summary:
+            Save file metadata to local file.
+        """
+
+        makedirs(dirname(file_loc), exist_ok=True)
+        with open(file_loc, 'w') as f:
+            json.dump(metadata, f, indent=4)
+
+    def download_file_metadata(self) -> List[Dict[str, Any]]:
+        """
+        Summary:
+            Download file metadata from metadata service, including.
+        Returns:
+            item_res (Dict[str, Any]): general metadata of file.
+            attribute_detail (Dict[str, Any]): attribute metadata of file.
+            tags (List[str]): tags metadata of file.
+        """
+
+        project_code, object_path = self.file_path.split('/', 1)
+        item_res = search_item(project_code, self.zone, object_path, 'file').get('result', {})
+        extra_info = item_res.pop('extended', {}).get('extra')
+        tags = extra_info.get('tags', [])
+        attributes = extra_info.get('attributes', [])
+        # use the uuid of attribute template to get template name
+        template_uuid = next(iter(attributes))
+        attribute_name = get_attribute_template_by_id(template_uuid).get('name')
+        attribute_detail = attributes.get(template_uuid)
+
+        # save metadata into files
+        self.save_file_metadata(self.general_location, item_res)
+        self.save_file_metadata(self.attribute_location, {attribute_name: attribute_detail})
+        self.save_file_metadata(self.tag_location, tags)
+
+        return item_res, {attribute_name: attribute_detail}, tags

app/services/output_manager/error_handler.py

@@ -63,6 +63,9 @@ class ECustomizedError(enum.Enum):
     INVALID_NAMEFOLDER = 'INVALID_NAMEFOLDER'
     INVALID_DOWNLOAD = 'INVALID_DOWNLOAD'
     DUPLICATE_TAG_ERROR = 'DUPLICATE_TAG_ERROR'
+    # file metadata related error
+    LOCAL_METADATA_FILE_EXISTS = 'LOCAL_METADATA_FILE_EXISTS'
+
     VERSION_NOT_EXIST = 'VERSION_NOT_EXIST'
     DATASET_NOT_EXIST = 'DATASET_NOT_EXIST'
     DATASET_PERMISSION = 'DATASET_PERMISSION'

app/services/output_manager/help_page.py

@@ -73,6 +73,12 @@ class FileHELP(enum.Enum):
     FILE_UPLOAD_PIPELINE = 'FILE_UPLOAD_PIPELINE'
     FILE_UPLOAD_ZIP = 'FILE_UPLOAD_ZIP'
 
+    FILE_META = 'FILE_META'
+    FILE_META_Z = 'FILE_META_Z'
+    FILE_META_G = 'FILE_META_G'
+    FILE_META_A = 'FILE_META_A'
+    FILE_META_T = 'FILE_META_T'
+
 
 def file_help_page(FileHELP: FileHELP):
     helps = help_msg.get('file', 'default file help')

app/services/output_manager/message_handler.py

@@ -195,6 +195,14 @@ def start_uploading(filename):
     def cancel_upload():
         logger.warning('Upload cancelled.')
 
+    @staticmethod
+    def cancel_metadata_download():
+        logger.warning('Metadata download cancelled.')
+
+    @staticmethod
+    def metadata_download_success():
+        logger.succeed('Metadata download complete.')
+
     @staticmethod
     def start_requests():
         """e.g. start requests."""

app/utils/aggregated.py

@@ -6,6 +6,7 @@
 import re
 import shutil
 from typing import Any
+from typing import Dict
 from typing import List
 
 import httpx
@@ -49,6 +50,18 @@ def search_item(project_code, zone, folder_relative_path, item_type, container_t
     return res.json()
 
 
+@require_valid_token()
+def get_attribute_template_by_id(template_id: str) -> Dict[str, Any]:
+    token = UserConfig().access_token
+    url = AppConfig.Connections.url_portal + f'/v1/data/manifest/{template_id}'
+    headers = {'Authorization': 'Bearer ' + token}
+    res = resilient_session().get(url, headers=headers)
+    if res.status_code != 200:
+        SrvErrorHandler.default_handle(res.text, True)
+
+    return res.json().get('result', {})
+
+
 @require_valid_token()
 def get_file_info_by_geid(geid: list):
     token = UserConfig().access_token

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "app"
-version = "2.8.6"
+version = "2.9.0"
 description = "This service is designed to support pilot platform"
 authors = ["Indoc Systems"]
 

tests/app/commands/test_entry_point.py

@@ -11,6 +11,7 @@
 from app.commands.file import file_download
 from app.commands.file import file_export_manifest
 from app.commands.file import file_list
+from app.commands.file import file_metadata_download
 from app.commands.file import file_put
 from app.commands.file import file_resume
 from app.commands.project import project_list_all
@@ -62,6 +63,7 @@ def test_file_commands(user_login_true):
         'attribute-export': file_export_manifest,
         'download': file_download,
         'resume': file_resume,
+        'metadata': file_metadata_download,
     }
     file_commands_object = entry_point.commands.get('file')
     file_commands_object.callback()