Skip to content

locationutils

locationutils

Module to access description for a remote storage location

Classes

LocationConfig

Access description for a remote storage location. The description is targeted at fsspec_.

Functions
__str__ 
__str__()

Returns string representation without credential values

Source code in niceml/utilities/fsspec/locationutils.py 
33
34
35
36
37
def __str__(self) -> str:
    """Returns string representation without credential values"""
    info = asdict(self)
    info["credentials"] = list(info["credentials"])
    return str(info)

Functions

get_location_uri 

get_location_uri(location)

Returns the URI of a LocationConfig.

Source code in niceml/utilities/fsspec/locationutils.py 
65
66
67
68
69
70
71
72
def get_location_uri(location: Union[LocationConfig, dict]) -> str:
    """Returns the URI of a LocationConfig."""
    parsed_config = (
        location
        if isinstance(location, LocationConfig)
        else cattr.structure(location, LocationConfig)
    )
    return parsed_config.uri

join_fs_path 

join_fs_path(file_system, *paths)

Returns joined given paths with the fsspec specific path seperator

Source code in niceml/utilities/fsspec/locationutils.py 
59
60
61
62
def join_fs_path(file_system: AbstractFileSystem, *paths: str) -> str:
    """Returns joined given paths with the fsspec specific path seperator"""
    paths = [path for path in paths if len(path) > 0]
    return file_system.sep.join(paths)

join_location_w_path 

join_location_w_path(location, path)

Returns joined LocationConfig with one or more path objects

Source code in niceml/utilities/fsspec/locationutils.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def join_location_w_path(
    location: Union[LocationConfig, dict], path: Union[List[str], str]
) -> LocationConfig:
    """Returns joined LocationConfig with one or more path objects"""
    parsed_config = (
        location
        if isinstance(location, LocationConfig)
        else cattr.structure(location, LocationConfig)
    )
    copied_config = deepcopy(parsed_config)
    # TODO: check how to get the correct separator from the filesystem
    if isinstance(path, list):
        for path_obj in path:
            copied_config.uri = join(copied_config.uri, path_obj)
    else:
        copied_config.uri = join(copied_config.uri, path)
    return copied_config

open_location 

open_location(config)

Creates a filesystem and path from configuration as a single context manager. The filesystem is "closed" (i.e. open connections are closed) when the context is left.

Source code in niceml/utilities/fsspec/locationutils.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@contextmanager
def open_location(config: Union[LocationConfig, Dict[str, Any]]) -> Iterator[FSPath]:
    """
    Creates a filesystem and path from configuration as a single context manager.
    The filesystem is "closed" (i.e. open connections are closed) when the context is left.
    """
    parsed_config = (
        config
        if isinstance(config, LocationConfig)
        else cattr.structure(config, LocationConfig)
    )
    credentials = deepcopy(parsed_config.credentials)
    fs_args = deepcopy(parsed_config.fs_args)
    filesystem, path = url_to_fs(parsed_config.uri, **credentials, **fs_args)
    try:
        yield filesystem, path
    finally:
        del filesystem