def download_azure_blob()

in container-filetransfer/cloud-transfer.py [0:0]

• 28 lines of code
• 6 McCabe index (conditional complexity)

def download_azure_blob(url, path):
    components = urlparse(url)
    query_string = parse_qs(components.query)
    if components.scheme in ['http', 'https']:
        if 'sig' not in query_string:
            logger.warning(f"[azure-blob] Skipping download of '{url}' because of incorrect scheme or missing SAS token")
            return False

        match = re.match(r'(\w+)\.blob\.(.+)', components.hostname)
        if not match:
            logger.warning(f"[azure-blob] Skipping upload to '{url}' because expected a hostname in format of 'accountname.blob.endpoint_suffix' (e.g. accountname.blob.core.windows.net).")
            return False
        account_name, endpoint_suffix = match.groups()

        # mount the storage so pycosio can detect the URL as backed by Azure
        parameters = {
            'account_name': account_name,
            'endpoint_suffix': endpoint_suffix,
            'sas_token': components.query
        }
        pycosio.mount(storage='azure_blob', storage_parameters=parameters)

        # remove query string from blob_url, pycosio will think it's part of the filename
        blob_url = urljoin(url, urlparse(url).path)

        path_components = urlparse(url).path.strip('/').split('/', 1)
        if len(path_components) == 2 and path_components[1]:
            pycosio.copyfile(blob_url, os.path.abspath(path))
        else:
            logger.warning(f"[azure-blob] Skipping download of '{url}' a specific blob must be specified; operating on whole blob containers is not yet supported.")
            return False
    else:
        logger.warning(f"[azure-blob] Skipping download of '{url}' because of incorrect scheme")
        return False