flowpipe.utilities module

Utilities for serializing and importing Nodes.

class flowpipe.utilities.NodeEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: JSONEncoder

Custom JSONEncoder to handle non-json serializable node values.

If the value is not json serializable, a sha256 hash of its bytes is encoded instead.

default(o: Any) → Any

Encode the object, handling type errors by encoding into sha256.

flowpipe.utilities.deserialize_graph(data: dict) → Graph

De-serialize from the given json data.

flowpipe.utilities.deserialize_node(data: dict) → INode

De-serialize a node from the given json data.

flowpipe.utilities.get_hash(obj: Any, hash_func: Callable = <function <lambda>>) → str | None

Safely get the hash of an object.

This function tries to compute the hash as safely as possible, dealing with json data and strings in a well-defined manner.

Parameters:

• obj – The object to hash

• hash_func (func(obj) -> str) – The hashing function to use

Returns:

A hash of the obj

Return type:

(str)

flowpipe.utilities.import_class(module_name: str, cls_name: str, file_location: str | None = None) → Type[Any]

Import and return the given class from the given module.

File location can be given to import the class from a location that is not accessible through the PYTHONPATH.

flowpipe.utilities.sanitize_string_input(input_str: str) → str

Escapes dangerous “{” for f strings. Call it before running format :param input_str: string to be sanitized :type input_str: str

Returns:

Sanitized string

Return type:

(str)