Skip to content

fsspecstorage

fsspecstorage

Module for fsspec storage

Classes

FSSpecStorage 

FSSpecStorage(fsconfig)

Bases: StorageInterface

A CloudStorageInterface to interact with fsspec isntances

Creates a new FSSpecStorage instance.

Parameters:

  • fsconfig (FSSpecStorage | Dict[str, Any]) –

    The fsspec configuration to open the filesystem with

Source code in niceml/data/storages/fsspecstorage.py 
26
27
28
29
30
31
32
33
34
def __init__(self, fsconfig: Union[LocationConfig, Dict[str, Any]]):
    """
    Creates a new FSSpecStorage instance.

    Args:
        fsconfig (FSSpecStorage | Dict[str, Any]):
            The fsspec configuration to open the filesystem with
    """
    self._fsconfig = fsconfig
Functions
download_as_str 
download_as_str(bucket_path)

returns the given bucket_path content as string

Raises:

  • RuntimeError –

    If the given bucket_path is not part of the currently opended filesystem.

Source code in niceml/data/storages/fsspecstorage.py 
65
66
67
68
69
70
71
72
73
74
def download_as_str(self, bucket_path: str) -> str:
    """returns the given bucket_path content as string

    Raises:
        RuntimeError:
            If the given bucket_path is not part of the currently opended
            filesystem.
    """
    with open_location(self._fsconfig) as (filesystem, path):
        return filesystem.cat(self.join_paths(path, bucket_path))
download_data 
download_data(bucket_path, local_path, recursive=True)

downloads a given object to the specified local_path

Raises:

  • RuntimeError –

    If the given bucket_path is not part of the currently opended filesystem.

Source code in niceml/data/storages/fsspecstorage.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def download_data(self, bucket_path: str, local_path: str, recursive: bool = True):
    """downloads a given object to the specified local_path

    Raises:
        RuntimeError:
            If the given bucket_path is not part of the currently opended
            filesystem.
    """
    with open_location(self._fsconfig) as (filesystem, path):
        local_dir = dirname(local_path)
        if not isdir(local_dir):
            makedirs(local_dir)
        filesystem.download(
            self.join_paths(path, bucket_path), local_path, recursive=recursive
        )
join_paths 
join_paths(*paths)

joins the given paths with the fsspec specific path seperator

Source code in niceml/data/storages/fsspecstorage.py 
76
77
78
79
80
def join_paths(self, *paths: str) -> str:
    """joins the given paths with the fsspec specific path seperator"""
    paths = [path for path in paths if len(path) > 0]
    with open_location(self._fsconfig) as (filesystem, _):
        return filesystem.sep.join(paths)
list_data 
list_data(path=None)

recusively lists all objects in the given path

Source code in niceml/data/storages/fsspecstorage.py 
36
37
38
39
40
41
42
43
44
45
46
47
def list_data(self, path: Optional[str] = None) -> List[str]:
    """recusively lists all objects in the given path"""
    with open_location(self._fsconfig) as (filesystem, fspath):
        target_path = fspath if path is None else self.join_paths(fspath, path)
        item_list = list_dir(
            target_path,
            return_full_path=True,
            recursive=True,
            file_system=filesystem,
        )
        item_list = [relpath(cur_file, fspath) for cur_file in item_list]
    return item_list

Functions