def walk_container()

in ai4e_azure_utils.py [0:0]

• 37 lines of code
• 12 McCabe index (conditional complexity)

def walk_container(container_client: ContainerClient,
                   max_depth: int = -1,
                   prefix: str = '',
                   store_folders: bool = True,
                   store_blobs: bool = True,
                   debug_max_items: int = -1) -> Tuple[List[str], List[str]]:
    """
    Recursively walk folders a Azure Blob Storage container.

    Based on:
    https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/storage/azure-storage-blob/samples/blob_samples_walk_blob_hierarchy.py
    """
    depth = 1

    def walk_blob_hierarchy(prefix: str,
                            folders: Optional[List[str]] = None,
                            blobs: Optional[List[str]] = None
                            ) -> Tuple[List[str], List[str]]:
        if folders is None:
            folders = []
        if blobs is None:
            blobs = []

        nonlocal depth

        if 0 < max_depth < depth:
            return folders, blobs

        for item in container_client.walk_blobs(name_starts_with=prefix):
            short_name = item.name[len(prefix):]
            if isinstance(item, BlobPrefix):
                # print('F: ' + prefix + short_name)
                if store_folders:
                    folders.append(prefix + short_name)
                depth += 1
                walk_blob_hierarchy(item.name, folders=folders, blobs=blobs)
                if (debug_max_items > 0
                        and len(folders) + len(blobs) > debug_max_items):
                    return folders, blobs
                depth -= 1
            else:
                if store_blobs:
                    blobs.append(prefix + short_name)

        return folders, blobs

    folders, blobs = walk_blob_hierarchy(prefix=prefix)

    assert all(s.endswith('/') for s in folders)
    folders = [s.strip('/') for s in folders]

    return folders, blobs