Skip to main content

utils

Utility functions for Spock

vars_dict_non_dunder#

def vars_dict_non_dunder(__obj: object)

Gets the user defined attributes from a base object class

Arguments:

  • __obj - class object to inspect for attribute values

Returns:

dictionary of non dunder attributes

make_salt#

def make_salt(salt_len: int = 16)

Make a salt of specific length

Arguments:

  • salt_len - length of the constructed salt

Returns:

salt string

_get_alias_type#

def _get_alias_type()

Gets the correct type of GenericAlias for versions less than 3.6

Returns:

_GenericAlias type

_get_callable_type#

def _get_callable_type()

Gets the correct underlying type reference for callable objects depending on the python version

Returns:

_VariadicGenericAlias type

_filter_optional#

def _filter_optional(val: List, allow_optional: bool = True)

Filters an iterable for None values if they are allowed

Arguments:

  • val - iterable of values that might contain None
  • allow_optional - allows the check to succeed if a given val in the iterable is None

Returns:

filtered list of values with None values removed

Raises:

_SpockValueError

sum_vals#

def sum_vals(val: List[Union[float, int, None]], sum_val: Union[float, int], allow_optional: bool = True, rel_tol: float = 1e-9, abs_tol: float = 0.0)

Checks if an iterable of values sums within tolerance to a specified value

Arguments:

  • val - iterable of values to sum
  • sum_val - sum value to compare against
  • allow_optional - allows the check to succeed if a given val in the iterable is None
  • rel_tol - relative tolerance – it is the maximum allowed difference between a and b
  • abs_tol - the minimum absolute tolerance – useful for comparisons near zero

Returns:

None

Raises:

_SpockValueError

eq_len#

def eq_len(val: List[Union[Tuple, List, None]], allow_optional: bool = True)

Checks that all values passed in the iterable are of the same length

Arguments:

  • val - iterable to compare lengths
  • allow_optional - allows the check to succeed if a given val in the iterable is None

Returns:

None

Raises:

_SpockValueError

within#

def within(val: Union[float, int, None], low_bound: Union[float, int], upper_bound: Union[float, int], inclusive_lower: bool = False, inclusive_upper: bool = False, allow_optional: bool = True) -> None

Checks that a value is within a defined range

Arguments:

  • val - value to check against
  • low_bound - lower bound of range
  • upper_bound - upper bound of range
  • inclusive_lower - if the check includes the bound value (i.e. >=)
  • inclusive_upper - if the check includes the bound value (i.e. <=)
  • allow_optional - allows the check to succeed if val is none

Returns:

None

Raises:

_SpockValueError

ge#

def ge(val: Union[float, int, None], bound: Union[float, int], allow_optional: bool = True) -> None

Checks that a value is greater than or equal to (inclusive) a lower bound

Arguments:

  • val - value to check against
  • bound - lower bound
  • allow_optional - allows the check to succeed if val is none

Returns:

None

Raises:

_SpockValueError

gt#

def gt(val: Union[float, int, None], bound: Union[float, int], allow_optional: bool = True) -> None

Checks that a value is greater (non-inclusive) than a lower bound

Arguments:

  • val - value to check against
  • bound - lower bound
  • allow_optional - allows the check to succeed if val is none

Returns:

None

Raises:

_SpockValueError

le#

def le(val: Union[float, int, None], bound: Union[float, int], allow_optional: bool = True) -> None

Checks that a value is less than or equal to (inclusive) an upper bound

Arguments:

  • val - value to check against
  • bound - upper bound
  • allow_optional - allows the check to succeed if val is none

Returns:

None

Raises:

_SpockValueError

lt#

def lt(val: Union[float, int], bound: Union[float, int], allow_optional: bool = True) -> None

Checks that a value is less (non-inclusive) than an upper bound

Arguments:

  • val - value to check against
  • bound - upper bound
  • allow_optional - allows the check to succeed if val is none

Returns:

None

Raises:

_SpockValueError

_find_all_spock_classes#

def _find_all_spock_classes(attr_class: _C) -> List

Within a spock class determine if there are any references to other spock classes

Arguments:

  • attr_class - a class with attrs attributes

Returns:

list of dependent spock classes

_check_4_spock_iterable#

def _check_4_spock_iterable(iter_obj: Union[Tuple, List]) -> bool

Checks if an iterable type contains a spock class

Arguments:

  • iter_obj - iterable type

Returns:

boolean if the iterable contains at least one spock class

_get_enum_classes#

def _get_enum_classes(enum_obj: EnumMeta) -> List

Checks if any of the values of an enum are spock classes and adds to a list

Arguments:

  • enum_obj - enum class

Returns:

list of enum values that are spock classes

path_object_to_s3path#

def path_object_to_s3path(path: Path) -> str

Convert a path object into a string s3 path

Arguments:

  • path - a spock config path

Returns:

string of s3 path

check_path_s3#

def check_path_s3(path: Path) -> bool

Checks the given path to see if it matches the s3:// regex

Arguments:

  • path - a spock config path

Returns:

boolean of regex match

_is_spock_instance#

def _is_spock_instance(__obj: object) -> bool

Checks if the object is a @spock decorated class

Private interface that checks to see if the object passed in is registered within the spock module and also is a class with attrs attributes (attrs_attrs)

Arguments:

  • __obj - class to inspect

Returns:

bool

_is_spock_instance_type#

def _is_spock_instance_type(__obj: object) -> bool

Checks if the object is a @spock decorated class type

Private interface that checks to see if the object passed in is registered within the spock module and also is a class with attrs attributes (attrs_attrs) and is of type (aka not instantiated)

Arguments:

  • __obj - class to inspect

Returns:

bool

_is_spock_tune_instance#

def _is_spock_tune_instance(__obj: object) -> bool

Checks if the object is a @spock decorated class

Private interface that checks to see if the object passed in is registered within the spock module tune addon and also is a class with attrs attributes (attrs_attrs)

Arguments:

  • __obj - class to inspect

Returns:

bool

_check_iterable#

def _check_iterable(iter_obj: Union[Tuple, List, EnumMeta]) -> bool

Check if an iterable type contains a spock class

Arguments:

  • iter_obj - iterable type

Returns:

boolean if the iterable contains at least one spock class

make_argument#

def make_argument(arg_name: str, arg_type: _T, parser: Type[_ArgumentGroup]) -> _ArgumentGroup

Make argparser argument based on type

Based on the type passed in handle the creation of the argparser argument so that overrides will have the correct behavior when set

Arguments:

  • arg_name - name for the argument
  • arg_type - type of the argument
  • parser - current parser

Returns:

  • parser - updated argparser

_handle_generic_type_args#

def _handle_generic_type_args(val: str) -> Any

Evaluates a string containing a Python literal

Seeing a list and tuple types will come in as string literal format, use ast to get the actual type

Arguments:

  • val - string literal

Returns:

the underlying string literal type

get_packages#

def get_packages() -> Dict

Gets all currently installed packages and assembles a dictionary of name: version

Notes:

https://stackoverflow.com/a/50013400

Returns:

dictionary of all currently available packages

add_info#

def add_info() -> Dict

Adds extra information to the output dictionary

Arguments:

Returns:

  • out_dict - output dictionary

make_blank_git#

def make_blank_git(out_dict: Dict) -> Dict

Adds blank git info

Arguments:

  • out_dict - current output dictionary

Returns:

  • out_dict - output dictionary with added git info

add_repo_info#

def add_repo_info(out_dict: Dict) -> Dict

Adds GIT information to the output dictionary

Arguments:

  • out_dict - output dictionary

Returns:

  • out_dict - output dictionary

add_generic_info#

def add_generic_info(out_dict: Dict) -> Dict

Adds date, fqdn information to the output dictionary

Arguments:

  • out_dict - output dictionary

Returns:

  • out_dict - output dictionary

_maybe_docker#

def _maybe_docker(cgroup_path: str = "/proc/self/cgroup") -> bool

Make a best effort to determine if run in a docker container

Arguments:

  • cgroup_path - path to cgroup file

Returns:

boolean of best effort docker determination

_maybe_k8s#

def _maybe_k8s(cgroup_path: str = "/proc/self/cgroup") -> bool

Make a best effort to determine if run in a container via k8s

Arguments:

  • cgroup_path - path to cgroup file

Returns:

boolean of best effort k8s determination

deep_payload_update#

def deep_payload_update(source: Dict, updates: Dict) -> Dict

Deeply updates a dictionary

Iterates through a dictionary recursively to update individual values within a possibly nested dictionary of dictionaries -- creates a dictionary if empty and trying to recurse

Arguments:

  • source - source dictionary
  • updates - updates to the dictionary

Returns:

  • source - updated version of the source dictionary

check_payload_overwrite#

def check_payload_overwrite(payload: Dict, updates: Dict, configs: str, overwrite: str = "") -> None

Warns when parameters are overwritten across payloads as order will matter

Arguments:

  • payload - current payload
  • payload_update - update to add to payload
  • configs - config path
  • overwrite - name of parent