Utilities and Miscellaneous#

exception UserReportableError[source]#

Bases: RuntimeError

An error reportable to the end user

Exceptions of this class or derived classes are able to be reported to the end user. Many exceptions should only be viewable by an administrator as they may leak implementation details or other sensitive information. Errors of this class are safe to report to all kinds of non-admin users.

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception UserManagementError[source]#

Bases: UserReportableError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception AuthenticationFailure[source]#

Bases: UserReportableError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception AuthorizationFailure[source]#

Bases: UserReportableError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception InvalidUsernameError[source]#

Bases: UserManagementError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception InvalidRolenameError[source]#

Bases: UserManagementError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception InvalidGroupnameError[source]#

Bases: UserManagementError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception InconsistentUpdateError[source]#

Bases: UserManagementError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception SecurityNotEnabledError[source]#

Bases: UserManagementError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception InvalidPasswordError[source]#

Bases: UserManagementError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception MissingDataError[source]#

Bases: UserReportableError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception CorruptDataError[source]#

Bases: UserReportableError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception AlreadyExistsError[source]#

Bases: UserReportableError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception LimitExceededError[source]#

Bases: UserReportableError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception InvalidUpdateError[source]#

Bases: UserReportableError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception NoClientError[source]#

Bases: RuntimeError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception ComputeManagerError[source]#

Bases: UserReportableError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception InvalidArgumentsError[source]#

Bases: UserReportableError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception SingleFileRequiredError[source]#

Bases: UserReportableError

__init__(*args, **kwargs)#

add_note()#: Exception.add_note(note) – add a note to the exception

args#

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class CompressionEnum[source]#

Bases: str, Enum

How data is compressed (compression method only, ie lzma, zstd)

none = 'none'#

lzma = 'lzma'#

zstd = 'zstd'#

__init__(*args, **kwds)#

get_compressed_ext(compression_type)[source]#

Parameters:: compression_type (str)
Return type:: str

compress(input_data, compression_type=CompressionEnum.zstd, compression_level=None)[source]#

Serializes and compresses data given a compression scheme and level

If compression_level is None, but a compression_type is specified, an appropriate default level is chosen

Returns a tuple containing the compressed data, applied compression type, and compression level (which may be different from the provided arguments)

Parameters:

input_data (Any)
compression_type (CompressionEnum)
compression_level (int | None)

Return type:

Tuple[bytes, CompressionEnum, int]

decompress(compressed_data, compression_type)[source]#

Decompresses and deserializes data into python objects

Parameters:

compressed_data (bytes)
compression_type (CompressionEnum)

Return type:

Any

deserialize(data: bytes | str, content_type: str, model: Type[_V]) → _V[source]#

deserialize(data: bytes | str, content_type: str, model: Any) → Any

Parameters:

data (bytes | str)
content_type (str)
model (Any)

Return type:

Any

serialize(data, content_type)[source]#

Parameters:: content_type (str)
Return type:: bytes

convert_numpy_recursive(obj, flatten=False)[source]#

is_scalar(obj)[source]#

Returns True if obj represents a single value rather than a collection of values

This is the rule used by make_list, and by functions that accept either a single value or a collection of values (and whose return value has a matching shape). With the exception of None, is_scalar(obj) is True exactly when make_list(obj) wraps obj in a new list instead of expanding it.

Anything sized and iterable (a list, tuple, set, frozenset, range, numpy array, or a view such as dict.keys()) is a collection of values. Strings and mappings are not - they are iterable, but are always treated as single values. Neither are objects that are merely iterable without being sized, such as generators or pydantic models.

None is a special case. It is neither a scalar nor a collection - make_list passes it through unchanged - and is_scalar(None) is True.

Parameters:: obj (Any)
Return type:: bool

make_list(obj: None) → None[source]#

make_list(obj: _S) → list[_S]

make_list(obj: _M) → list[_M]

make_list(obj: Set[_T]) → list[_T]

make_list(obj: Sequence[_T]) → list[_T]

make_list(obj: Any) → list[Any]

Returns a list of the values in obj, or a list containing obj if it is a single value

See is_scalar for what counts as a single value. Sets, numpy arrays, and views such as dict.keys() are all expanded into a list. None is passed through unchanged.

Parameters:: obj (Any)
Return type:: Any

chunk_iterable(it, chunk_size)[source]#

Split an iterable (such as a list) into batches/chunks

Parameters:

it (Iterable[_T])
chunk_size (int)

Return type:

Generator[list[_T], None, None]

chunk_iterable_time(it, chunk_time, max_chunk_size, initial_chunk_size)[source]#

Split an iterable into chunks, trying to keep a constant time per chunk

This function keeps track of the time it takes to process each chunk and tries to keep the time per chunk as close to ‘chunk_time’ as possible, increasing or decreasing the chunk size as needed (up to ‘max_chunk_size’)

The first chunk will be of size ‘initial_chunk_size’ (assuming there is enough elements in the iterable to fill it).

Parameters:

it (Iterable[_T])
chunk_time (float)
max_chunk_size (int)
initial_chunk_size (int)

Return type:

Generator[list[_T], None, None]

process_chunk_iterable(fn, it, chunk_time, max_chunk_size, initial_chunk_size, max_workers=1, *, keep_order=False)[source]#

Process an iterable in chunks, trying to keep a constant time per chunk

This function keeps track of the time it takes to process each chunk and tries to keep the time per chunk as close to ‘chunk_time’ as possible, increasing or decreasing the chunk size as needed (up to ‘max_chunk_size’)

The first chunk will be of size ‘initial_chunk_size’ (assuming there is enough elements in the iterable to fill it).

This function yields whatever fn returned for each chunk. If ‘keep_order’ is True, the results will be returned in the same order as the original iterable. If ‘keep_order’ is False, the results will be returned in the order they are completed.

Parameters:

fn (Callable[[list[_T]], _U])
it (Iterable[_T])
chunk_time (float)
max_chunk_size (int)
initial_chunk_size (int)
max_workers (int)
keep_order (bool)

Return type:

Generator[_U, None, None]

process_iterable(fn, it, chunk_time, max_chunk_size, initial_chunk_size, max_workers=1, *, keep_order=False)[source]#

Similar to process_chunk_iterable, but returns individual elements rather than chunks

Parameters:

fn (Callable[[list[_T]], Iterable[_U]])
it (Iterable[_T])
chunk_time (float)
max_chunk_size (int)
initial_chunk_size (int)
max_workers (int)
keep_order (bool)

Return type:

Generator[_U, None, None]

seconds_to_hms(seconds)[source]#

Converts a number of seconds (as an integer) to a string representing hh:mm:ss

Parameters:: seconds (float | int)
Return type:: str

duration_to_seconds(s)[source]#

Parses a string in dd:hh:mm:ss or 1d2h3m4s to an integer number of seconds

Parameters:: s (int | str | float)
Return type:: int

recursive_normalizer(value, digits=10, lowercase=True)[source]#

Prepare a structure for hashing by lowercasing all values and round all floats

Parameters:

value (Any)
digits (int)
lowercase (bool)

Return type:

Any

calculate_limit(max_limit, given_limit)[source]#

Get the allowed limit on results to return for a particular or type of object

If ‘given_limit’ is given (ie, by the user), this will return min(limit, max_limit) where max_limit is the set value for the table/type of object

Parameters:

max_limit (int)
given_limit (int | None)

Return type:

int

hash_dict(d)[source]#

Parameters:: d (dict[str, Any])
Return type:: str

reshape_molecule(a)[source]#

Converts a flattened list with length N to a nested list of dimensions (N,3)

Parameters:: a (list)
Return type:: list[list[float]]

capture_all_output(top_logger)[source]#

Captures all output, including stdout, stderr, and logging

Parameters:: top_logger (str)

now_at_utc()[source]#

Get the current time as a timezone-aware datetime object

Return type:: datetime

is_included(key, include, exclude, default)[source]#

Determine if a field should be included given the include and exclude lists

Handles “*” and “**” as well

Parameters:

key (str)
include (Iterable[str] | None)
exclude (Iterable[str] | None)
default (bool)

Return type:

bool

update_nested_dict(d, u)[source]#

Parameters:

d (dict[str, Any])
u (dict[str, Any])

apply_jitter(t, jitter_fraction)[source]#

Parameters:

t (int | float)
jitter_fraction (float)

Return type:

float

time_based_cache(seconds=10, maxsize=None)[source]#

Parameters:

seconds (int)
maxsize (int | None)