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: TDict,
copy_src_dicts: bool = False) -> TDict
Merges src into dst key wise. Does not recur into lists. Values in src overwrite dst if both keys exit.
Only dict and its subclasses are updated recursively. With copy_src_dicts, dict key:values will be deep copied,
otherwise, both dst and src will keep the same references.
clone_dict_nested
def clone_dict_nested(src: TDict) -> TDict
Clones src structure descending into nested dicts. Does not descend into mappings that are not dicts ie. specs instances.
Compared to deepcopy does not clone any other objects. Uses update_dict_nested internally
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],
normalize_f: Callable[[str], str] = str.__call__) -> 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.