Source code for galaxy.tool_util.verify.wait

"""Abstraction for waiting on API conditions to become true."""

import time
from typing import (
    Callable,
    Optional,
    Union,
)

DEFAULT_POLLING_BACKOFF = 0
DEFAULT_POLLING_DELTA = 0.25

TIMEOUT_MESSAGE_TEMPLATE = "Timed out after {} seconds waiting on {}."
timeout_type = Union[int, float]


[docs]def wait_on(
    function: Callable,
    desc: str,
    timeout: timeout_type,
    delta: timeout_type = DEFAULT_POLLING_DELTA,
    polling_backoff: timeout_type = DEFAULT_POLLING_BACKOFF,
    sleep_: Optional[Callable] = None,
):
    """Wait for function to return non-None value.

    Grow the polling interval (initially ``delta`` defaulting to 0.25 seconds)
    incrementally by the supplied ``polling_backoff`` (defaulting to 0).

    Throw a TimeoutAssertionError if the supplied timeout is reached without
    supplied function ever returning a non-None value.
    """
    sleep = sleep_ or time.sleep
    total_wait = 0.0
    while True:
        if total_wait > timeout:
            raise TimeoutAssertionError(TIMEOUT_MESSAGE_TEMPLATE.format(total_wait, desc))
        value = function()
        if value is not None:
            return value
        total_wait += delta
        sleep(delta)
        delta += polling_backoff


[docs]class TimeoutAssertionError(AssertionError):
    """Derivative of AssertionError indicating wait_on exceeded max time."""

[docs]    def __init__(self, message):
        super().__init__(message)