#!/usr/bin/env python
from azure_storage.methods import copy_blob, create_parent_parser, delete_container, delete_file, delete_folder, \
extract_common_path, move_prep, setup_arguments
from argparse import ArgumentParser, RawTextHelpFormatter
import coloredlogs
import logging
import sys
import os
class AzureContainerMove(object):
def main(self):
# Validate the container names, and prepare all the necessary clients
self.container_name, self.target_container, self.blob_service_client, self.source_container_client, \
self.target_container_client = move_prep(
passphrase=self.passphrase,
account_name=self.account_name,
container_name=self.container_name,
target_container=self.target_container)
# Rename (move) the container
self.move_container(source_container_client=self.source_container_client,
blob_service_client=self.blob_service_client,
container_name=self.container_name,
target_container=self.target_container,
path=self.path,
storage_tier=self.storage_tier)
# Delete the original container once the copy is complete
delete_container(blob_service_client=self.blob_service_client,
container_name=self.container_name,
account_name=self.account_name)
@staticmethod
def move_container(source_container_client, blob_service_client, container_name, target_container, path,
storage_tier):
"""
Rename (move) the specified container in Azure storage
:param source_container_client: type azure.storage.blob.BlobServiceClient.ContainerClient for source container
:param blob_service_client: type: azure.storage.blob.BlobServiceClient
:param container_name: type str: Name of the container in which the folder is located
:param target_container: type str: Name of the container into which the folder is to be moved
:param path: type str: Path of folders in which the files are to be placed
:param storage_tier: type str: Storage tier to use for the container
"""
# Create a generator containing all the blobs in the container
generator = source_container_client.list_blobs()
for blob_file in generator:
# Copy the file to the new container
copy_blob(blob_file=blob_file,
blob_service_client=blob_service_client,
container_name=container_name,
target_container=target_container,
path=path,
storage_tier=storage_tier,
category='container')
def __init__(self, container_name, account_name, passphrase, target_container, path, 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.target_container = target_container
self.path = path
self.storage_tier = storage_tier
self.connect_str = str()
self.blob_service_client = None
self.source_container_client = None
self.target_container_client = None
class AzureMove(object):
def main(self):
# Validate the container names, and prepare all the necessary clients
self.container_name, self.target_container, self.blob_service_client, self.source_container_client, \
self.target_container_client = move_prep(
passphrase=self.passphrase,
account_name=self.account_name,
container_name=self.container_name,
target_container=self.target_container)
# Run the proper method depending on whether a file or a folder is requested
if self.category == 'file':
self.move_file(source_container_client=self.source_container_client,
object_name=self.object_name,
blob_service_client=self.blob_service_client,
container_name=self.container_name,
target_container=self.target_container,
path=self.path,
storage_tier=self.storage_tier)
delete_file(container_client=self.source_container_client,
object_name=self.object_name,
blob_service_client=self.blob_service_client,
container_name=self.container_name)
elif self.category == 'folder':
self.move_folder(source_container_client=self.source_container_client,
object_name=self.object_name,
blob_service_client=self.blob_service_client,
container_name=self.container_name,
target_container=self.target_container,
path=self.path,
category=self.category,
storage_tier=self.storage_tier)
delete_folder(container_client=self.source_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
@staticmethod
def move_file(source_container_client, object_name, blob_service_client, container_name, target_container,
path, storage_tier):
"""
Move the specified file to the desired container in Azure storage
:param source_container_client: type azure.storage.blob.BlobServiceClient.ContainerClient for source container
:param object_name: type str: Name and path of folder to move in Azure storage
:param blob_service_client: type: azure.storage.blob.BlobServiceClient
:param container_name: type str: Name of the container in which the folder is located
:param target_container: type str: Name of the container into which the folder is to be moved
:param path: type str: Path of folders in which the files are to be placed
:param storage_tier: type str: Storage tier to use for the file
"""
# Create a generator containing all the blobs in the container
generator = source_container_client.list_blobs()
# Create a boolean to determine if the blob has been located
present = False
for blob_file in generator:
# Filter for the blob name
if blob_file.name == object_name:
# Update the blob presence variable
present = True
# Copy the file to the new container
copy_blob(blob_file=blob_file,
blob_service_client=blob_service_client,
container_name=container_name,
target_container=target_container,
path=path,
storage_tier=storage_tier)
# Send a warning to the user that the blob could not be found
if not present:
logging.error(f'Could not locate the desired file {object_name}')
raise SystemExit
@staticmethod
def move_folder(source_container_client, object_name, blob_service_client, container_name, target_container, path,
storage_tier, category):
"""
Move the specified folder (and its contents) to the desired container in Azure storage
:param source_container_client: type azure.storage.blob.BlobServiceClient.ContainerClient for source container
:param object_name: type str: Name and path of folder to move in Azure storage
:param blob_service_client: type: azure.storage.blob.BlobServiceClient
:param container_name: type str: Name of the container in which the folder is located
:param target_container: type str: Name of the container into which the folder is to be moved
:param path: type str: Path of folders in which the files are to be placed
:param storage_tier: type str: Storage tier to use for the moved folder
:param category: type str: Category of object to be copied. Limited to file or folder
"""
# Create a generator containing all the blobs in the container
generator = source_container_client.list_blobs()
# Create a boolean to determine if the blob has been located
present = False
for blob_file in generator:
# Extract the common path between the current file and the requested folder
common_path = extract_common_path(object_name=object_name,
blob_file=blob_file)
# Only copy the file if there is a common path between the object path and the blob path (they match)
if common_path is not None:
# Update the blob presence variable
present = True
# Copy the file to the new container
copy_blob(blob_file=blob_file,
blob_service_client=blob_service_client,
container_name=container_name,
target_container=target_container,
path=path,
object_name=object_name,
category=category,
common_path=common_path,
storage_tier=storage_tier)
# Send a warning to the user that the blob could not be found
if not present:
logging.error(f'Could not locate the desired folder {object_name}')
raise SystemExit
def __init__(self, object_name, container_name, account_name, passphrase, target_container, path,
storage_tier, 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.target_container = target_container
self.path = path
self.storage_tier = storage_tier
self.category = category
self.connect_str = str()
self.blob_service_client = None
self.source_container_client = None
self.target_container_client = None
def container_move(args):
"""
Run the AzureContainerMove method
:param args: type ArgumentParser arguments
"""
logging.info(f'Renaming container {args.container_name} to {args.target_container} in Azure storage '
f'account {args.account_name}')
move_container = AzureContainerMove(container_name=args.container_name,
account_name=args.account_name,
passphrase=args.passphrase,
target_container=args.target_container,
path=args.reset_path,
storage_tier=args.storage_tier)
move_container.main()
def file_move(args):
"""
Run the AzureMove method for a file
:param args: type ArgumentParser arguments
"""
logging.info(f'Moving file {args.file} from {args.container_name} to {args.target_container} in Azure storage '
f'account {args.account_name}')
move_file = AzureMove(object_name=args.file,
container_name=args.container_name,
account_name=args.account_name,
passphrase=args.passphrase,
target_container=args.target_container,
path=args.reset_path,
storage_tier=args.storage_tier,
category='file')
move_file.main()
def folder_move(args):
"""
Run the AzureMove method for a folder
:param args: type ArgumentParser arguments
"""
logging.info(f'Moving folder {args.folder} from {args.container_name} to {args.target_container} in Azure storage '
f'account {args.account_name}')
move_folder = AzureMove(object_name=args.folder,
container_name=args.container_name,
account_name=args.account_name,
passphrase=args.passphrase,
target_container=args.target_container,
path=args.reset_path,
storage_tier=args.storage_tier,
category='folder')
move_folder.main()
def cli():
parser = ArgumentParser(description='Move containers, files, or folders in Azure storage')
# Create the parental parser, and the subparser
subparsers, parent_parser = create_parent_parser(parser=parser)
parent_parser.add_argument('-t', '--target_container',
required=True,
help='The target container to which the container/file/folder is to be moved '
'(this can be the same as the container_name if you want to move a file/folder'
' within a container')
parent_parser.add_argument('-r', '--reset_path',
type=str,
help='Set the path of the container/file/folder within a folder in the target container '
'e.g. sequence_data/220202-m05722. If you want to place it directly in the '
'container without any nesting, use or \'\'')
parent_parser.add_argument('-s', '--storage_tier',
type=str,
default='Hot',
choices=['Hot', 'Cool', 'Archive'],
metavar='STORAGE_TIER',
help='Set the storage tier for the container/file/folder to be moved. '
'Options are "Hot", "Cool", and "Archive". Default is Hot')
# Container move subparser
container_move_subparser = subparsers.add_parser(parents=[parent_parser],
name='container',
description='Move a container in Azure storage',
formatter_class=RawTextHelpFormatter,
help='Move a container in Azure storage')
container_move_subparser.set_defaults(func=container_move)
# File move subparser
file_move_subparser = subparsers.add_parser(parents=[parent_parser],
name='file',
description='Move a file within Azure storage',
formatter_class=RawTextHelpFormatter,
help='Move a file within Azure storage')
file_move_subparser.add_argument('-f', '--file',
type=str,
required=True,
help='Name of blob file to move in Azure storage. '
'e.g. 2022-SEQ-0001_S1_L001_R1_001.fastq.gz')
file_move_subparser.set_defaults(func=file_move)
# Folder move subparser
folder_move_subparser = subparsers.add_parser(parents=[parent_parser],
name='folder',
description='Move a folder within Azure storage',
formatter_class=RawTextHelpFormatter,
help='Move a folder within Azure storage')
folder_move_subparser.add_argument('-f', '--folder',
type=str,
required=True,
help='Name of folder to move in Azure storage. '
'e.g. InterOp')
folder_move_subparser.set_defaults(func=folder_move)
# 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('Move 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()