utilities

Module with useful functions

• successive_pairs

following_pairs_second_item(ell, search_pair)

Used to get affected table names

Source code in xerini/utilities.py

def following_pairs_second_item(
    ell: List[Tuple[Things, Things]], search_pair: Tuple[Things, Things]
) -> Things:
    """Used to get affected table names"""
    if search_pair not in ell:
        raise ValueError(f"The {search_pair=} is not in {ell=}!")
    idx: int = ell.index(search_pair)
    if (idx + 1) == len(ell):
        raise IndexError(f"{search_pair} was the last element of {ell=}!")
    return ell[idx + 1][1]

meaningful_strings_count(text=None)

We need this function to determine whether the text used to initialize a SQL Statement only has
a single statement
:param text: an arbitrary text
:param separator: space, tab, but most likely ;
:return: the number of meaningful string in the original text

Source code in xerini/utilities.py

def meaningful_strings_count(text: Optional[str] = None) -> int:
    """
    We need this function to determine whether the text used to initialize a SQL Statement only has
    a single statement
    :param text: an arbitrary text
    :param separator: space, tab, but most likely ;
    :return: the number of meaningful string in the original text
    """
    ans = 0
    if text is None:
        pass
    if text:
        ans = sum(bool(sub.replace(";", "")) for sub in sqlparse.split(text))
    return ans

minimal_nodes(directed)

Returns the set of nodes in the network that have no incoming edges
:param directed: the graph from which we should extract the minimal nodes
:return: a set of mininal nodes

Source code in xerini/utilities.py

def minimal_nodes(directed: nx.DiGraph) -> set:
    """
    Returns the set of nodes in the network that have no incoming edges
    :param directed: the graph from which we should extract the minimal nodes
    :return: a set of mininal nodes
    """
    if not isinstance(directed, nx.DiGraph):
        raise TypeError(f"unable to compute minimal nodes for {type(directed)=}.")
    return {node for node in directed.nodes() if directed.in_degree(node) == 0}

parallel_decomposition(directed)

Decompose the DAG into stages

Source code in xerini/utilities.py

def parallel_decomposition(directed: nx.DiGraph) -> Iterator[set]:
    """Decompose the DAG into stages"""
    if not nx.is_directed_acyclic_graph(directed):
        raise TypeError("parallel decomposition requires a directed acyclic graph.")
    _ctdg = directed.copy()
    while _ctdg.nodes():
        minimals = minimal_nodes(_ctdg)
        _ctdg.remove_nodes_from(minimals)
        yield minimals

random_id8()

Returns an ascii string of length 8 which starts with a lower case character
:return:

Source code in xerini/utilities.py

def random_id8() -> str:
    """
    Returns an ascii string of length 8 which starts with a lower case character
    :return:
    """
    alpha_numeric = ascii_lowercase + digits
    first = choice(ascii_lowercase)
    rest = "".join(choices(alpha_numeric, k=7))
    return first + rest

random_partitions(number_of_partitions, items)

Partitions the elements of items into a tuple of random lists of roughly equal size

Source code in xerini/utilities.py

def random_partitions(
    number_of_partitions, items: Collection[Things]
) -> Tuple[List[Things], ...]:
    """Partitions the elements of items into a tuple of random lists of roughly equal size"""
    if len(items) < number_of_partitions:
        raise ValueError(
            f"The provided collection has less elements, {len(items)=}, "
            f"than the number of partitions, {number_of_partitions=}!"
        )
    if number_of_partitions < 2:
        raise ValueError(
            f"The partition size, {number_of_partitions=}, should be greater than 2!"
        )
    _t: Tuple[List[Things], ...] = tuple([] for _ in range(number_of_partitions))
    _ell = list(items)
    shuffle(_ell)
    for index, elem in enumerate(_ell):
        _t[index % number_of_partitions].append(elem)
    return _t

successive_pairs(iterable)

Given an iterable this function returns a generator of the successive overlapping
pairs taken from the input iterable,

>>> list(successive_pairs('ABCDEFG'))
[('A', 'B'), ('B', 'C'), ('C', 'D'), ('D', 'E'), ('E', 'F'), ('F', 'G')]

Source code in xerini/utilities.py

def successive_pairs(iterable: Iterable[Things]) -> Iterable[Tuple[Things, Things]]:
    """
    Given an iterable this function returns a generator of the successive overlapping
    pairs taken from the input iterable,
    ```python
    >>> list(successive_pairs('ABCDEFG'))
    [('A', 'B'), ('B', 'C'), ('C', 'D'), ('D', 'E'), ('E', 'F'), ('F', 'G')]
    ```
    """
    _a, _b = tee(iterable)
    next(_b, None)
    return zip(_a, _b)