Skip to content

fsfilesystemstorage

fsfilesystemstorage

Module for fsspec storage

Classes

FsFileSystemStorage 

FsFileSystemStorage(file_system, root_dir)

Bases: StorageInterface

A CloudStorageInterface to interact with fsspec isntances

Creates a new FSSpecStorage instance.

Source code in niceml/data/storages/fsfilesystemstorage.py 
22
23
24
25
26
27
def __init__(self, file_system: AbstractFileSystem, root_dir: str):
    """
    Creates a new FSSpecStorage instance.
    """
    self.file_system = file_system
    self.root_dir = root_dir
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/fsfilesystemstorage.py 
58
59
60
61
62
63
64
65
66
67
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.
    """

    return self.file_system.cat(self.join_paths(self.root_dir, 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/fsfilesystemstorage.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
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.
    """
    local_dir = dirname(local_path)
    if not isdir(local_dir):
        makedirs(local_dir)
    self.file_system.download(
        self.join_paths(self.root_dir, 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/fsfilesystemstorage.py 
69
70
71
72
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]
    return self.file_system.sep.join(paths)
list_data 
list_data(path=None)

recusively lists all objects in the given path

Source code in niceml/data/storages/fsfilesystemstorage.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
def list_data(self, path: Optional[str] = None) -> List[str]:
    """recusively lists all objects in the given path"""
    target_path = (
        self.root_dir if path is None else self.join_paths(self.root_dir, path)
    )
    item_list = list_dir(
        target_path,
        return_full_path=True,
        recursive=True,
        file_system=self.file_system,
    )
    item_list = [relpath(cur_file, self.root_dir) for cur_file in item_list]
    return item_list

Functions