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.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: str, Enum

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

none = 'none'#
lzma = 'lzma'#
zstd = 'zstd'#
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:
Return type:

Any

encode_to_json(obj, encoder=None)[source]#

Takes an object and turns it into plain python that can be encoded to JSON.

This does not actually turn the object into JSON (string), just prepares it to be done. This is useful for turning various objects into something that can be put into a JSON(B) column in the database

Parameters:

  • obj (Any)

  • encoder (JSONEncoder | None)

Return type:

Any

deserialize(data, content_type)[source]#
Parameters:

  • data (bytes | str)

  • content_type (str)

serialize(data, content_type)[source]#
Parameters:

content_type (str)

Return type:

bytes

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

Returns a list containing obj if obj is not a list or other iterable type object

This will also work with sets

Parameters:

obj (_T | Sequence[_T] | Set[_T] | None)

Return type:

List[_T] | None

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 returns the results as chunks (lists) of the original iterable. 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[[Iterable[_T]], Any])

  • it (Iterable[_T])

  • chunk_time (float)

  • max_chunk_size (int)

  • initial_chunk_size (int)

  • max_workers (int)

  • keep_order (bool)

Return type:

Generator[List[_T], 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 ranther than chunks

Parameters:

  • fn (Callable[[Iterable[_T]], Any])

  • it (Iterable[_T])

  • chunk_time (float)

  • max_chunk_size (int)

  • initial_chunk_size (int)

  • max_workers (int)

  • keep_order (bool)

Return type:

Generator[List[_T], 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

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

On this page

This Page