import os from argparse import Namespace, _SubParsersAction from mdrsclient.api import FolderApi from mdrsclient.commands.base import BaseCommand from mdrsclient.commands.utils import ( create_connection, find_folder, find_laboratory, parse_remote_host_with_path, ) class FolderCommand(BaseCommand): @staticmethod def register(top_level_subparsers: _SubParsersAction) -> None: # ls ls_parser = top_level_subparsers.add_parser("ls", help="list the folder contents") ls_parser.add_argument("remote_path", help="Remote folder path (remote:/lab/path/)") ls_parser.set_defaults(func=FolderCommand.list) # mkdir mkdir_parser = top_level_subparsers.add_parser("mkdir", help="create a new folder") mkdir_parser.add_argument("remote_path", help="Remote folder path (remote:/lab/path/)") mkdir_parser.set_defaults(func=FolderCommand.mkdir) # rmdir rmdir_parser = top_level_subparsers.add_parser("rmdir", help="remove a existing folder") rmdir_parser.add_argument("remote_path", help="Remote folder path (remote:/lab/path/)") rmdir_parser.set_defaults(func=FolderCommand.rmdir) # metadata metadata_parser = top_level_subparsers.add_parser("metadata", help="get a folder metadata") metadata_parser.add_argument("remote_path", help="Remote folder path (remote:/lab/path/)") metadata_parser.set_defaults(func=FolderCommand.metadata) @staticmethod def list(args: Namespace) -> None: (remote, laboratory_name, path) = parse_remote_host_with_path(args.remote_path) connection = create_connection(remote) laboratory = find_laboratory(connection, laboratory_name) folder = find_folder(connection, laboratory, path) label = { "type": "Type", "acl": "Access", "laboratory": "Laboratory", "size": "Lock/Size", "date": "Date", "name": "Name", } length: dict[str, int] = {} for key in label.keys(): length[key] = len(label[key]) for sub_folder in folder.sub_folders: sub_laboratory = connection.laboratories.find_by_id(sub_folder.lab_id) sub_laboratory_name = sub_laboratory.name if sub_laboratory is not None else "(invalid)" length["acl"] = max(length["acl"], len(sub_folder.access_level_name)) length["laboratory"] = max(length["laboratory"], len(sub_laboratory_name)) length["size"] = max(length["size"], len(sub_folder.lock_name)) length["date"] = max(length["date"], len(sub_folder.updated_at_name)) length["name"] = max(length["name"], len(sub_folder.name)) for file in folder.files: length["size"] = max(length["size"], len(str(file.size))) length["date"] = max(length["date"], len(file.updated_at_name)) length["name"] = max(length["name"], len(file.name)) length["acl"] = max(length["acl"], len(folder.access_level_name)) length["laboratory"] = max(length["laboratory"], len(laboratory.name)) header = ( f"{label['type']:{length['type']}}\t{label['acl']:{length['acl']}}\t" f"{label['laboratory']:{length['laboratory']}}\t{label['size']:{length['size']}}\t" f"{label['date']:{length['date']}}\t{label['name']:{length['name']}}" ) print(header) print("-" * len(header.expandtabs())) for sub_folder in sorted(folder.sub_folders, key=lambda x: x.name): sub_laboratory = connection.laboratories.find_by_id(sub_folder.lab_id) sub_laboratory_name = sub_laboratory.name if sub_laboratory is not None else "(invalid)" print( f"{'[d]':{length['type']}}\t{sub_folder.access_level_name:{length['acl']}}\t" f"{sub_laboratory_name:{length['laboratory']}}\t{sub_folder.lock_name:{length['size']}}\t" f"{sub_folder.updated_at_name:{length['date']}}\t{sub_folder.name:{length['name']}}" ) for file in sorted(folder.files, key=lambda x: x.name): print( f"{'[f]':{length['type']}}\t{folder.access_level_name:{length['acl']}}\t" f"{laboratory.name:{length['laboratory']}}\t{file.size:{length['size']}}\t" f"{file.updated_at_name:{length['date']}}\t{file.name:{length['name']}}" ) @staticmethod def mkdir(args: Namespace) -> None: (remote, laboratory_name, path) = parse_remote_host_with_path(args.remote_path) path = path.rstrip("/") parent_path = os.path.dirname(path) folder_name = os.path.basename(path) connection = create_connection(remote) laboratory = find_laboratory(connection, laboratory_name) folder = find_folder(connection, laboratory, parent_path) folder_api = FolderApi(connection) folder_api.create(folder_name, folder.id) @staticmethod def rmdir(args: Namespace) -> None: (remote, laboratory_name, path) = parse_remote_host_with_path(args.remote_path) connection = create_connection(remote) laboratory = find_laboratory(connection, laboratory_name) folder = find_folder(connection, laboratory, path) folder_api = FolderApi(connection) folder_api.destroy(folder.id) @staticmethod def metadata(args: Namespace) -> None: (remote, laboratory_name, path) = parse_remote_host_with_path(args.remote_path) connection = create_connection(remote) laboratory = find_laboratory(connection, laboratory_name) folder = find_folder(connection, laboratory, path) folder_api = FolderApi(connection) metadata = folder_api.metadata(folder.id) print(metadata)