hats.io.file_io.file_pointer#

Attributes#

Functions#

get_upath(→ upath.UPath)

Returns a UPath file pointer from a path string or other path-like type.

get_upath_for_protocol(→ upath.UPath)

Create UPath with protocol-specific configurations.

filter_query_params_for_url(→ dict)

Return a copy of query_params with any keys unsupported by the target

directory_has_contents(→ bool)

Checks if a directory already has some contents (any files or subdirectories)

Module Contents#

BLOCK_SIZE = 32768[source]#
get_upath(path: str | pathlib.Path | upath.UPath) upath.UPath[source]#

Returns a UPath file pointer from a path string or other path-like type.

Parameters:
path: str | Path | UPath

base file path to be normalized to UPath

Returns:
UPath

Instance of UPath.

get_upath_for_protocol(path: str | pathlib.Path) upath.UPath[source]#

Create UPath with protocol-specific configurations.

If we access pointers on S3 and credentials are not found we assume an anonymous access, i.e., that the bucket is public.

Parameters:
path: str | Path | UPath

base file path to be normalized to UPath

Returns:
UPath

Instance of UPath.

filter_query_params_for_url(url: str, query_params: dict) dict[source]#

Return a copy of query_params with any keys unsupported by the target host removed.

Currently the only special case is VizCaT (vizcat.cds.unistra.fr), which does not support server-side filters.

Parameters:
urlstr

Full URL of the catalog resource (used only to inspect the host).

query_paramsdict

Query parameters dict as produced by dict_to_query_urlparams().

Returns:
dict

Filtered copy of query_params.

directory_has_contents(pointer: str | pathlib.Path | upath.UPath) bool[source]#

Checks if a directory already has some contents (any files or subdirectories)

Parameters:
pointerstr | Path | UPath

File Pointer to check for existing contents

Returns:
bool

True if there are any files or subdirectories below this directory.