common.utils
uniq_id
def uniq_id(len_: int = 16) -> str
Returns a hex encoded crypto-grade string of random bytes with desired len_
uniq_id_base64
def uniq_id_base64(len_: int = 16) -> str
Returns a base64 encoded crypto-grade string of random bytes with desired len_
many_uniq_ids_base64
def many_uniq_ids_base64(n_ids: int, len_: int = 16) -> List[str]
Generate n_ids base64 encoded crypto-grade strings of random bytes with desired len_.
This is more performant than calling uniq_id_base64 multiple times.
digest128
def digest128(v: str, len_: int = 15) -> str
Returns a base64 encoded shake128 hash of str v with digest of length len_ (default: 15 bytes = 20 characters length)
digest128b
def digest128b(v: bytes, len_: int = 15) -> str
Returns a base64 encoded shake128 hash of bytes v with digest of length len_ (default: 15 bytes = 20 characters length)
flatten_list_of_str_or_dicts
def flatten_list_of_str_or_dicts(
seq: Sequence[Union[StrAny, str]]) -> DictStrAny
Transforms a list of objects or strings [{K: {...}}, L, ...] -> {K: {...}, L: None, ...}
concat_strings_with_limit
def concat_strings_with_limit(strings: List[str], separator: str,
limit: int) -> Iterator[str]
Generator function to concatenate strings.
The function takes a list of strings and concatenates them into a single string such that the length of each concatenated string does not exceed a specified limit. It yields each concatenated string as it is created. The strings are separated by a specified separator.
Arguments:
stringsList[str] - The list of strings to be concatenated.separatorstr - The separator to use between strings. Defaults to a single space.limitint - The maximum length for each concatenated string.
Yields:
Generator[str, None, None]: A generator that yields each concatenated string.
graph_edges_to_nodes
def graph_edges_to_nodes(edges: Sequence[Tuple[TAny, TAny]],
directed: bool = True) -> Dict[TAny, Set[TAny]]
Converts a directed graph represented as a sequence of edges to a graph represented as a mapping from nodes a set of connected nodes.
Isolated nodes are represented as edges to itself. If directed is False, each edge is duplicated but going in opposite direction.
graph_find_scc_nodes
def graph_find_scc_nodes(undag: Dict[TAny, Set[TAny]]) -> List[Set[TAny]]
Finds and returns a list of sets of nodes in strongly connected components of a undag which is undirected
To obtain undirected graph from edges use graph_edges_to_nodes function with directed argument False.
update_dict_with_prune
def update_dict_with_prune(dest: DictStrAny, update: StrAny) -> None
Updates values that are both in dest and update and deletes dest values that are None in update
update_dict_nested
def update_dict_nested(dst: TDict,
src: StrAny,
keep_dst_values: bool = False) -> TDict
Merges src into dst key wise. Does not recur into lists. Values in src overwrite dst if both keys exit.
Optionally (keep_dst_values) you can keep the dst value on conflict
map_nested_in_place
def map_nested_in_place(func: AnyFun, _complex: TAny) -> TAny
Applies func to all elements in _dict recursively, replacing elements in nested dictionaries and lists in place.
is_interactive
def is_interactive() -> bool
Determine if the current environment is interactive.
Returns:
bool- True if interactive (e.g., REPL, IPython, Jupyter Notebook), False if running as a script.
custom_environ
@contextmanager
def custom_environ(env: StrStr) -> Iterator[None]
Temporarily set environment variables inside the context manager and fully restore previous environment afterwards
multi_context_manager
@contextmanager
def multi_context_manager(
managers: Sequence[ContextManager[Any]]) -> Iterator[Any]
A context manager holding several other context managers. Enters and exists all of them. Yields from the last in the list
is_inner_callable
def is_inner_callable(f: AnyFun) -> bool
Checks if f is defined within other function
get_module_name
def get_module_name(m: ModuleType) -> str
Gets module name from module with a fallback for executing module main
derives_from_class_of_name
def derives_from_class_of_name(o: object, name: str) -> bool
Checks if object o has class of name in its derivation tree
compressed_b64encode
def compressed_b64encode(value: bytes) -> str
Compress and b64 encode the given bytestring
compressed_b64decode
def compressed_b64decode(value: str) -> bytes
Decode a bytestring encoded with compressed_b64encode
merge_row_counts
def merge_row_counts(row_counts_1: RowCounts, row_counts_2: RowCounts) -> None
merges row counts_2 into row_counts_1
extend_list_deduplicated
def extend_list_deduplicated(original_list: List[Any],
extending_list: Iterable[Any]) -> List[Any]
extends the first list by the second, but does not add duplicates
maybe_context
@contextmanager
def maybe_context(manager: ContextManager[TAny]) -> Iterator[TAny]
Allows context manager manager to be None by creating dummy context. Otherwise manager is used
without_none
def without_none(d: Mapping[TKey, Optional[TValue]]) -> Mapping[TKey, TValue]
Return a new dict with all None values removed
get_exception_trace
def get_exception_trace(exc: BaseException) -> ExceptionTrace
Get exception trace and additional information for DltException(s)
get_exception_trace_chain
def get_exception_trace_chain(exc: BaseException,
traces: List[ExceptionTrace] = None,
seen: Set[int] = None) -> List[ExceptionTrace]
Get traces for exception chain. The function will recursively visit all cause and context exceptions. The top level exception trace is first on the list
order_deduped
def order_deduped(lst: List[Any]) -> List[Any]
Returns deduplicated list preserving order of input elements.
Only works for lists with hashable elements.