vayu.common

common

retry

retry(
    exception_to_check: Union[_ExcType, Tuple[_ExcType]],
    tries: int = 4,
    delay: float = 1,
    backoff: float = 2,
    logger: Logger = None,
)

Retry calling the decorated function using an exponential backoff. Asyncio compatible.

Parameters:

Name	Type	Description	Default
exception_to_check	Union[_ExcType, Tuple[_ExcType]]	the exception to check. Maybe a tuple of exceptions to check.	required
tries	int	number of times to try (not retry) before giving up	4
delay	float	initial delay between retries in seconds	1
backoff	float	backoff multiplier e.g. value of 2 will double the delay for each retry	2
logger	Logger	logger to use. If None, print.	None

Source code in vayu/common.py

def __init__(
    self,
    exception_to_check: Union[_ExcType, Tuple[_ExcType]],
    tries: int = 4,
    delay: float = 1,
    backoff: float = 2,
    logger: Logger = None,
):
    """
    Args:
        exception_to_check: the exception to check. Maybe a tuple of exceptions to check.
        tries: number of times to try (not retry) before giving up
        delay: initial delay between retries in seconds
        backoff: backoff multiplier e.g. value of 2 will double the delay for each retry
        logger: logger to use. If None, print.
    """
    self._tries = tries
    self._delay = delay
    self._backoff = backoff
    self._ignore_exc_types = exception_to_check
    self._log_fn = logger.warning if logger else print

add_jitter

add_jitter(
    n: JitterT, jitter_percentage: int = 20
) -> JitterT

Add jitter to seconds.

Source code in vayu/common.py

def add_jitter(n: JitterT, jitter_percentage: int = 20) -> JitterT:
    """Add jitter to seconds."""
    return n * (1 + random.randint(-jitter_percentage, jitter_percentage) / 100)

group

group(iterable: Iterable, key: Callable) -> Dict

Group elements of an iterable by a key function.

Source code in vayu/common.py

def group(iterable: Iterable, key: Callable) -> Dict:
    """Group elements of an iterable by a key function."""
    groups = {}
    for item in iterable:
        k = key(item)
        if k not in groups:
            groups[k] = []
        groups[k].append(item)
    return groups