abs

abs

Classes

Functions

get_files_to_download

get_files_to_download(cloud_interface, source, dest)

Download a file or directory to a path on the local filesystem

Parameters:

• cloud_interface (StorageInterface) –

  interface to load data

• source (str) –

  Source path in the container

• dest (str) –

  Local destination path

Source code in niceml/storages/abs.py

def get_files_to_download(cloud_interface: StorageInterface, source: str, dest: str):
    """
    Download a file or directory to a path on the local filesystem

    Args:
        cloud_interface: interface to load data
        source: Source path in the container
        dest: Local destination path
    """
    logger = logging.getLogger("abs_logger")
    logger.disabled = True
    return list_download_files(cloud_interface, source, dest)

list_download_files

list_download_files(
    cloud_interface, bucket_path, local_path
)

List all files that need to be downloaded from the given source and the resulting destination paths. Args: cloud_interface: interface to load data bucket_path: Source path in the container local_path: Local destination path

Returns:

• Dict[str, str] –

  Dictionary with the blob path as key and the local path as value

Source code in niceml/storages/abs.py

def list_download_files(
    cloud_interface: StorageInterface, bucket_path: str, local_path: str
) -> Dict[str, str]:
    """
    List all files that need to be downloaded from the given source and
    the resulting destination paths.
    Args:
        cloud_interface: interface to load data
        bucket_path: Source path in the container
        local_path: Local destination path

    Returns:
        Dictionary with the blob path as key and the local path as value
    """
    blobs = cloud_interface.list_data(bucket_path)
    download_dict = {}
    if blobs:
        for blob_path in blobs:
            file_path = os.path.relpath(blob_path, bucket_path)
            if not file_path.startswith(local_path):
                file_path = os.path.join(local_path, file_path)
            if not os.path.exists(file_path):
                download_dict[blob_path] = file_path
    else:
        download_dict[bucket_path] = local_path
    return download_dict