#!/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()