Utils

filter_keys_in_settings(dictionary, settings_obj)

Utility function for checking the membership of dictionary keys in a settings class definition.

Parameters:

Name	Type	Description	Default
dictionary	dict	Dictionary to check	required
settings_obj	BaseSettings	Settings object for composing	required

Source code in lume_services/utils.py

def filter_keys_in_settings(dictionary: dict, settings_obj: BaseSettings) -> dict:
    """Utility function for checking the membership of dictionary keys in a settings
    class definition.

    Args:
        dictionary (dict): Dictionary to check
        settings_obj (BaseSettings): Settings object for composing

    """
    not_in_settings = [
        key for key in dictionary.keys() if key not in settings_obj.attributes
    ]
    in_settings = [
        key for key in dictionary.keys() if key not in settings_obj.attributes
    ]

    if len(not_in_settings):
        logger.warning(
            "Key %s not found in settings. Allowed keys are for %s are %s",
            ",".join(not_in_settings),
            settings_obj.class_name,
            ",".join(settings_obj.attributes),
        )

    return {key: value for key, value in dictionary.items() if key in in_settings}

fingerprint_dict(dictionary)

Create a hash for a dictionary

Parameters:

Name	Type	Description	Default
dictionary	dict	Dictionary for which to create a fingerprint hash.	required

Source code in lume_services/utils.py

def fingerprint_dict(dictionary: dict):
    """Create a hash for a dictionary

    Args:
        dictionary (dict): Dictionary for which to create a fingerprint hash.

    """

    dictionary = get_jsonable_dict(dictionary)

    hasher = hashlib.md5()
    hasher.update(json.dumps(dictionary).encode("utf-8"))
    return hasher.hexdigest()

flatten_dict_for_query(dictionary, level_key=None)

Flatten a dictionary of values for pymongo query

Source code in lume_services/utils.py

def flatten_dict_for_query(dictionary: dict, level_key: str = None):
    """Flatten a dictionary of values for pymongo query"""

    flattened_dict = {}

    for key, value in dictionary.items():

        if level_key is not None:
            key = f"{level_key}.{key}"

        if isinstance(value, dict):
            value_dict = flatten_dict_for_query(value, level_key=key)
            flattened_dict.update(value_dict)

        else:
            flattened_dict[key] = value

    return flattened_dict

get_callable_from_string(callable, bind=None)

Get callable from a string. In the case that the callable points to a bound method, the function returns a callable taking the bind instance as the first arg.

Parameters:

Name	Type	Description	Default
callable	str	String representation of callable abiding convention module:callable	required
bind	Any	Class to bind as self	None

Returns:

Type	Description
Callable	Callable

Source code in lume_services/utils.py

def get_callable_from_string(callable: str, bind: Any = None) -> Callable:
    """Get callable from a string. In the case that the callable points to a bound
    method, the function returns a callable taking the bind instance as the first arg.

    Args:
        callable: String representation of callable abiding convention
             __module__:callable
        bind: Class to bind as self

    Returns:
        Callable
    """
    callable_split = callable.rsplit(".", 1)

    if len(callable_split) != 2:
        raise ValueError(f"Improperly formatted callable string: {callable_split}")

    module_name, callable_name = callable_split

    try:
        module = import_module(module_name)

    except ModuleNotFoundError:

        try:
            module_split = module_name.rsplit(".", 1)

            if len(module_split) != 2:
                raise ValueError(f"Unable to access: {callable}")

            module_name, class_name = module_split

            module = import_module(module_name)
            callable_name = f"{class_name}.{callable_name}"

        except ModuleNotFoundError as err:
            logger.error("Unable to import module %s", module_name)
            raise err

        except ValueError as err:
            logger.error(err)
            raise err

    # construct partial in case of bound method
    if "." in callable_name:
        bound_class, callable_name = callable_name.rsplit(".")

        try:
            bound_class = getattr(module, bound_class)
        except Exception as e:
            logger.error("Unable to get %s from %s", bound_class, module_name)
            raise e

        # require right partial for assembly of callable
        # https://funcy.readthedocs.io/en/stable/funcs.html#rpartial
        def rpartial(func, *args):
            return lambda *a: func(*(a + args))

        callable = getattr(bound_class, callable_name)
        params = inspect.signature(callable).parameters

        # check bindings
        is_bound = params.get("self", None) is not None
        if not is_bound and bind is not None:
            raise ValueError("Cannot bind %s to %s.", callable_name, bind)

        # bound, return partial
        if bind is not None:
            if not isinstance(bind, (bound_class,)):
                raise ValueError(
                    "Provided bind %s is not instance of %s",
                    bind,
                    bound_class.__qualname__,
                )

        if is_bound and isinstance(callable, (FunctionType,)) and bind is None:
            callable = rpartial(getattr, callable_name)

        elif is_bound and isinstance(callable, (FunctionType,)) and bind is not None:
            callable = getattr(bind, callable_name)

    else:
        if bind is not None:
            raise ValueError("Cannot bind %s to %s.", callable_name, type(bind))

        try:
            callable = getattr(module, callable_name)
        except Exception as e:
            logger.error("Unable to get %s from %s", callable_name, module_name)
            raise e

    return callable

get_jsonable_dict(dictionary)

Converts numpy arrays inside a dictionary to list items.

Source code in lume_services/utils.py

def get_jsonable_dict(dictionary: dict):
    """Converts numpy arrays inside a dictionary to list items."""

    def convert_array_values(dictionary):
        """Convert array values to list so the dictionary is json serializable."""
        dictionary = {
            key: value.tolist() if isinstance(value, (np.ndarray,)) else value
            for key, value in dictionary.items()
        }
        # convert pandas dataframe to json
        dictionary = {
            key: value.to_json() if isinstance(value, (pd.DataFrame,)) else value
            for key, value in dictionary.items()
        }
        dictionary = {
            key: convert_array_values(value) if isinstance(value, (dict,)) else value
            for key, value in dictionary.items()
        }
        return dictionary

    return convert_array_values(dictionary)

select_python_version(version)

Utility for selecting a python version given a version string formatted with a pin.

Parameters:

Name	Type	Description	Default
version	str	String rep of version with pin	required

Returns:

Name	Type	Description
str	str	Sting rep of version to use

Raises:

Type	Description
ValueError	Unable to parse python version.

Source code in lume_services/utils.py

def select_python_version(version: str) -> str:
    """Utility for selecting a python version given a version string formatted with
    a pin.

    Args:
        version (str): String rep of version with pin

    Returns:
        str: Sting rep of version to use

    Raises:
        ValueError: Unable to parse python version.

    """

    # if >=, use base
    if ">=" in version or "<=" in version or "=" in version:
        v = float(version.split("=")[1])

    elif ">" in version:
        v = float(version.replace(">", "")) + 0.1

    elif "<" in version:
        v = float(version.replace("<", "")) - 0.1

    else:
        raise ValueError("Cannot parse python version %s", version)

    return str(v)