#!/usr/bin/env python from azure_storage.methods import client_prep, create_blob_client, create_parent_parser, setup_arguments from argparse import ArgumentParser, RawTextHelpFormatter import coloredlogs import logging import azure import sys import os class AzureContainerTier(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, create=False) self.container_tier(container_client=self.container_client, blob_service_client=self.blob_service_client, container_name=self.container_name, storage_tier=self.storage_tier) @staticmethod def container_tier(container_client, blob_service_client, container_name, storage_tier): """ Set the storage tier for the specified container in Azure storage :param container_client: type azure.storage.blob.BlobServiceClient.ContainerClient :param blob_service_client: type: azure.storage.blob.BlobServiceClient :param container_name: type str: Name of the container of interest :param storage_tier: type str: Storage tier to use for the container """ # Create a generator containing all the blobs in the container generator = container_client.list_blobs() try: # Hide the INFO-level messages sent to the logger from Azure by increasing the logging level to WARNING logging.getLogger().setLevel(logging.WARNING) for blob_file in generator: # Create the blob client blob_client = create_blob_client(blob_service_client=blob_service_client, container_name=container_name, blob_file=blob_file) # Set the storage tier blob_client.set_standard_blob_tier(standard_blob_tier=storage_tier) except azure.core.exceptions.ResourceNotFoundError: logging.error(f' The specified container, {container_name}, does not exist.') raise SystemExit def __init__(self, container_name, account_name, passphrase, storage_tier): # Set the container name variable self.container_name = container_name # Initialise necessary class variables self.passphrase = passphrase self.account_name = account_name self.storage_tier = storage_tier self.connect_str = str() self.blob_service_client = None self.container_client = None class AzureTier(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, create=False) # Run the proper method depending on whether a file or a folder is requested if self.category == 'file': self.file_tier(container_client=self.container_client, object_name=self.object_name, blob_service_client=self.blob_service_client, container_name=self.container_name, storage_tier=self.storage_tier) elif self.category == 'folder': self.folder_tier(container_client=self.container_client, object_name=self.object_name, blob_service_client=self.blob_service_client, container_name=self.container_name, storage_tier=self.storage_tier) else: logging.error(f'Something is wrong. There is no {self.category} option available') raise SystemExit @staticmethod def file_tier(container_client, object_name, blob_service_client, container_name, storage_tier): """ Set the storage tier for the specified file in Azure storage :param container_client: type azure.storage.blob.BlobServiceClient.ContainerClient :param object_name: type str: Name and path of file for which a SAS URL is to be created :param blob_service_client: type: azure.storage.blob.BlobServiceClient :param container_name: type str: Name of the container of interest :param storage_tier: type str: Storage tier to use for the file """ # Create a generator containing all the blobs in the container generator = container_client.list_blobs() # Create a boolean to determine if the blob has been located present = False # Hide the INFO-level messages sent to the logger from Azure by increasing the logging level to WARNING logging.getLogger().setLevel(logging.WARNING) try: for blob_file in generator: # Filter for the blob name if blob_file.name == object_name: # Update the blob presence variable present = True # Create the blob client blob_client = create_blob_client(blob_service_client=blob_service_client, container_name=container_name, blob_file=blob_file) # Set the storage tier blob_client.set_standard_blob_tier(standard_blob_tier=storage_tier) # Send an error to the user that the blob could not be found if not present: logging.error(f'Could not locate the desired file {object_name} in {container_name}') raise SystemExit except azure.core.exceptions.ResourceNotFoundError: logging.error(f' The specified container, {container_name}, does not exist.') raise SystemExit @staticmethod def folder_tier(container_client, object_name, blob_service_client, container_name, storage_tier): """ Set the storage tier for the specified folder in Azure storage :param container_client: type azure.storage.blob.BlobServiceClient.ContainerClient :param object_name: type str: Name and path of file for which a SAS URL is to be created :param blob_service_client: type: azure.storage.blob.BlobServiceClient :param container_name: type str: Name of the container of interest :param storage_tier: type str: Storage tier to use for the folder """ # Create a generator containing all the blobs in the container generator = container_client.list_blobs() # Create a boolean to determine if the blob has been located present = False # Hide the INFO-level messages sent to the logger from Azure by increasing the logging level to WARNING logging.getLogger().setLevel(logging.WARNING) try: for blob_file in generator: # Create the path of the file by extracting the path of the file blob_path = os.path.join(os.path.split(blob_file.name)[0]) # Ensure that the supplied folder path is present in the blob path if os.path.normpath(object_name) in os.path.normpath(blob_path): # Update the folder presence boolean present = True # Create the blob client blob_client = create_blob_client(blob_service_client=blob_service_client, container_name=container_name, blob_file=blob_file) # Set the storage tier blob_client.set_standard_blob_tier(standard_blob_tier=storage_tier) # Send an error to the user that the folder could not be found if not present: logging.error(f'Could not locate the desired folder {object_name} in container {container_name}') raise SystemExit except azure.core.exceptions.ResourceNotFoundError: logging.error(f' The specified container, {container_name}, does not exist.') raise SystemExit def __init__(self, object_name, container_name, account_name, passphrase, storage_tier, category): # Set the name of the file/folder to have its storage tier set 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.storage_tier = storage_tier self.category = category self.connect_str = str() self.blob_service_client = None self.container_client = None def container_tier(args): """ Run the AzureContainerTier method :param args: type ArgumentParser arguments """ logging.info(f'Setting the storage tier for Azure container {args.container_name} to {args.storage_tier}') # Create the container tier object container_tier_set = AzureContainerTier(container_name=args.container_name, account_name=args.account_name, passphrase=args.passphrase, storage_tier=args.storage_tier) container_tier_set.main() def file_tier(args): """ Run the AzureTier method for a file :param args: type ArgumentParser arguments """ logging.info(f'Setting the storage tier for file {args.file} in Azure container {args.container_name} to ' f'{args.storage_tier}') # Create the file tier object file_tier_set = AzureTier(container_name=args.container_name, object_name=args.file, account_name=args.account_name, passphrase=args.passphrase, storage_tier=args.storage_tier, category='file') file_tier_set.main() def folder_tier(args): """ Run the AzureTier method for a file :param args: type ArgumentParser arguments """ logging.info(f'Setting the storage tier for folder {args.folder} in Azure container {args.container_name} to ' f'{args.storage_tier}') # Create the folder tier object folder_tier_set = AzureTier(container_name=args.container_name, object_name=args.folder, account_name=args.account_name, passphrase=args.passphrase, storage_tier=args.storage_tier, category='folder') folder_tier_set.main() def cli(): parser = ArgumentParser(description='Set the storage tier of containers/files/folders in Azure storage') # Create the parental parser, and the subparser subparsers, parent_parser = create_parent_parser(parser=parser) parent_parser.add_argument('-s', '--storage_tier', type=str, required=True, choices=['Hot', 'Cool', 'Archive'], metavar='STORAGE_TIER', help='Set the storage tier for a container/file/folder. Options are "Hot", ' '"Cool", and "Archive"') # Container tier setting parser container_subparser = subparsers.add_parser(parents=[parent_parser], name='container', description='Change the storage tier of a container in Azure storage', formatter_class=RawTextHelpFormatter, help='Change the storage tier of a container in Azure storage') container_subparser.set_defaults(func=container_tier) # File tier setting parser file_subparser = subparsers.add_parser(parents=[parent_parser], name='file', description='Change the storage tier of a file in Azure storage', formatter_class=RawTextHelpFormatter, help='Change the storage tier of a file in Azure storage') file_subparser.add_argument('-f', '--file', type=str, required=True, help='Name of file in Azure storage that will have its storage tier set' 'e.g. 220202-m05722/2022-SEQ-0001_S1_L001_R1_001.fastq.gz') file_subparser.set_defaults(func=file_tier) # Folder downloading subparser folder_subparser = subparsers.add_parser(parents=[parent_parser], name='folder', description='Change the storage tier of a folder in Azure storage', formatter_class=RawTextHelpFormatter, help='Change the storage tier of a folder in Azure storage') folder_subparser.add_argument('-f', '--folder', type=str, required=True, help='Name of the folder in Azure storage that will have its storage tier set ' 'e.g. InterOp') folder_subparser.set_defaults(func=folder_tier) # 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('Storage tier set') # 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()