PrivacyScanData/Azure/AzureStorage/azure_storage/azure_delete.py

#!/usr/bin/env python
from azure_storage.methods import client_prep, create_parent_parser, delete_container, delete_file, \
    delete_folder, setup_arguments, set_blob_retention_policy
from argparse import ArgumentParser, RawTextHelpFormatter
import coloredlogs
import logging
import sys
import os


class AzureContainerDelete(object):

    def main(self):
        self.container_name, self.connect_str, self.blob_service_client, container_client = \
            client_prep(container_name=self.container_name,
                        passphrase=self.passphrase,
                        account_name=self.account_name)
        # Hide the INFO-level messages sent to the logger from Azure by increasing the logging level to WARNING
        logging.getLogger().setLevel(logging.WARNING)
        delete_container(blob_service_client=self.blob_service_client,
                         container_name=self.container_name,
                         account_name=self.account_name)

    def __init__(self, container_name, account_name, passphrase):
        # Set the container name variable
        self.container_name = container_name
        # Initialise necessary class variables
        self.passphrase = passphrase
        self.account_name = account_name
        self.connect_str = str()
        self.blob_service_client = None


class AzureDelete(object):

    def main(self):
        self.container_name, self.connect_str, self.blob_service_client, self.container_client = \
            client_prep(container_name=self.container_name,
                        passphrase=self.passphrase,
                        account_name=self.account_name)
        # Hide the INFO-level messages sent to the logger from Azure by increasing the logging level to WARNING
        logging.getLogger().setLevel(logging.WARNING)
        # Set the file retention policy
        self.blob_service_client = set_blob_retention_policy(blob_service_client=self.blob_service_client,
                                                             days=self.retention_time)
        # Run the proper method depending on whether a file or a folder is requested
        if self.category == 'file':
            delete_file(container_client=self.container_client,
                        object_name=self.object_name,
                        blob_service_client=self.blob_service_client,
                        container_name=self.container_name)
        elif self.category == 'folder':
            delete_folder(container_client=self.container_client,
                          object_name=self.object_name,
                          blob_service_client=self.blob_service_client,
                          container_name=self.container_name,
                          account_name=self.account_name)
        else:
            logging.error(f'Something is wrong. There is no {self.category} option available')
            raise SystemExit

    def __init__(self, object_name, container_name, account_name, passphrase, retention_time, category):
        self.object_name = object_name
        # Set the container name variable
        self.container_name = container_name
        # Initialise necessary class variables
        self.passphrase = passphrase
        self.account_name = account_name
        self.retention_time = retention_time
        # Ensure that the retention time provided is valid
        try:
            assert 0 < self.retention_time < 366
        except AssertionError:
            logging.error(f'The provided retention time ({self.retention_time}) is invalid. '
                          f'It must be between 1 and 365 days')
            raise SystemExit
        self.category = category
        self.connect_str = str()
        self.blob_service_client = None
        self.container_client = None


def container_delete(args):
    """
    Run the AzureContainerDelete method
    :param args: type ArgumentParser arguments
    """
    logging.info(f'Deleting container {args.container_name} from Azure storage account {args.account_name}')
    del_container = AzureContainerDelete(container_name=args.container_name,
                                         account_name=args.account_name,
                                         passphrase=args.passphrase)
    del_container.main()


def file_delete(args):
    """
    Run the AzureDelete method for a file
    :param args: type ArgumentParser arguments
    """
    logging.info(f'Deleting file {args.file} from container {args.container_name} in Azure storage account '
                 f'{args.account_name}')
    del_file = AzureDelete(object_name=args.file,
                           container_name=args.container_name,
                           account_name=args.account_name,
                           passphrase=args.passphrase,
                           retention_time=args.retention_time,
                           category='file')
    del_file.main()


def folder_delete(args):
    """
    Run the AzureDelete method for a folder
    :param args: type ArgumentParser arguments
    """
    logging.info(f'Deleting folder {args.folder} from container {args.container_name} in Azure storage account '
                 f'{args.account_name}')
    del_folder = AzureDelete(object_name=args.folder,
                             container_name=args.container_name,
                             account_name=args.account_name,
                             passphrase=args.passphrase,
                             retention_time=args.retention_time,
                             category='folder')
    del_folder.main()


def cli():
    parser = ArgumentParser(description='Move and/or delete containers, files, or folders in Azure storage')
    # Create the parental parser, and the subparser
    subparsers, parent_parser = create_parent_parser(parser=parser)
    # Container delete subparser
    container_delete_subparser = subparsers.add_parser(parents=[parent_parser],
                                                       name='container',
                                                       description='Delete a container in Azure storage',
                                                       formatter_class=RawTextHelpFormatter,
                                                       help='Delete a container in Azure storage')
    container_delete_subparser.set_defaults(func=container_delete)
    # File delete subparser
    file_delete_subparser = subparsers.add_parser(parents=[parent_parser],
                                                  name='file',
                                                  description='Delete a file in Azure storage',
                                                  formatter_class=RawTextHelpFormatter,
                                                  help='Delete a file in Azure storage')
    file_delete_subparser.add_argument('-f', '--file',
                                       type=str,
                                       required=True,
                                       help='Name of blob file to delete in Azure storage. '
                                            'e.g. 2022-SEQ-0001_S1_L001_R1_001.fastq.gz')
    file_delete_subparser.add_argument('-r', '--retention_time',
                                       type=int,
                                       default=8,
                                       help='Retention time for deleted files. Default is 8 days. Must be between '
                                            '1 and 365')
    file_delete_subparser.set_defaults(func=file_delete)
    # Folder delete subparser
    folder_delete_subparser = subparsers.add_parser(parents=[parent_parser],
                                                    name='folder',
                                                    description='Delete a folder in Azure storage',
                                                    formatter_class=RawTextHelpFormatter,
                                                    help='Delete a folder in Azure storage')
    folder_delete_subparser.add_argument('-f', '--folder',
                                         type=str,
                                         required=True,
                                         help='Name of folder to delete in Azure storage. '
                                              'e.g. InterOp')
    folder_delete_subparser.add_argument('-r', '--retention_time',
                                         type=int,
                                         default=8,
                                         help='Retention time for deleted files. Default is 8 days')
    folder_delete_subparser.set_defaults(func=folder_delete)
    # Set up the arguments, and run the appropriate subparser
    arguments = setup_arguments(parser=parser)
    # Return to the requested logging level, as it has been increased to WARNING to suppress the log being filled with
    # information from azure.core.pipeline.policies.http_logging_policy
    coloredlogs.install(level=arguments.verbosity.upper())
    logging.info('Deletion complete')
    # Prevent the arguments being printed to the console (they are returned in order for the tests to work)
    sys.stderr = open(os.devnull, 'w')
    return arguments


if __name__ == '__main__':
    cli()