pytest_wdl/utils.py
def poll(
    target: Callable,
    step: int = 1,
    args: Optional[Sequence] = None,
    kwargs: Optional[dict] = None,
    timeout: Optional[int] = None,
    max_tries: Optional[int] = None,
    check_success: Callable = bool,
    step_function: Optional[Callable[[int, int], int]] = None,
    ignore_exceptions: Sequence = (),
):
    """
    Poll by calling a target function until a certain condition is met. You must specify
    at least a target function to be called and the step -- base wait time between
    each function call.

    Vendored from the [polling](https://github.com/justiniso/polling) package.

    Args:
        target: The target callable
        step: Step defines the amount of time to wait (in seconds)
        args: Arguments to be passed to the target function
        kwargs: Keyword arguments to be passed to the target function
        timeout: The target function will be called until the time elapsed is greater
            than the maximum timeout (in seconds). NOTE that the actual execution
            time of the function *can* exceed the time specified in the timeout. For
            instance, if the target function takes 10 seconds to execute and the timeout
            is 21 seconds, the polling function will take a total of 30 seconds (two
            iterations of the target --20s which is less than the timeout--21s,
            and a final iteration)
        max_tries: Maximum number of times the target function will be called before
            failing
        check_success: A callback function that accepts the return value of the target
            function. It must return true if you want the polling function to stop
            and return this value. It must return false if you want to continue
            polling. You may also use this function to collect non-success values. The
            default is a callback that tests for truthiness (anything not False, 0,
            or empty collection).
        step_function: A callback function that accepts two arguments: current_step,
            num_tries; and returns the next step value. By default, this is constant,
            but you can also pass a function that will increase or decrease the step.
            As an example, you can increase the wait time between calling the target
            function by 10 seconds every iteration until the step is 100 seconds--at
            which point it should remain constant at 100 seconds

            >>> def my_step_function(current_step: int, num_tries: int) -> int:
            >>>     return max(current_step + 10, 100)

        ignore_exceptions: You can specify a tuple of exceptions that should be caught
            and ignored on every iteration. If the target function raises one of
            these exceptions, it will be caught and the exception instance will be
            pushed to the queue of values collected during polling. Any other exceptions
            raised will be raised as normal.

    Returns:
        The first value from the target function that meets the condions of the
        check_success callback. By default, this will be the first value that is not
        None, 0, False, '', or an empty collection.
    """
    max_time = time.time() + timeout if timeout else None
    tries = 0
    last_item = None

    if args is None:
        args = ()

    if kwargs is None:
        kwargs = {}

    while True:
        if max_tries and tries >= max_tries:
            raise MaxCallException(last_item)

        try:
            val = target(*args, **kwargs)
            last_item = val
        except ignore_exceptions as e:
            last_item = e
        else:
            # Condition passes, this is the only "successful" exit from the
            # polling function
            if check_success(val):
                return val

        tries += 1

        # Check the time after to make sure the poll function is called at least once
        if max_time and time.time() >= max_time:
            raise TimeoutException(last_item)

        time.sleep(step)

        if step_function:
            step = step_function(step, tries)

    Poll by calling a target function until a certain condition is met. You must specify
    at least a target function to be called and the step -- base wait time between
    each function call.

    Vendored from the [polling](https://github.com/justiniso/polling) package.

    Args:
        target: The target callable
        step: Step defines the amount of time to wait (in seconds)
        args: Arguments to be passed to the target function
        kwargs: Keyword arguments to be passed to the target function
        timeout: The target function will be called until the time elapsed is greater
            than the maximum timeout (in seconds). NOTE that the actual execution
            time of the function *can* exceed the time specified in the timeout. For
            instance, if the target function takes 10 seconds to execute and the timeout
            is 21 seconds, the polling function will take a total of 30 seconds (two
            iterations of the target --20s which is less than the timeout--21s,
            and a final iteration)
        max_tries: Maximum number of times the target function will be called before
            failing
        check_success: A callback function that accepts the return value of the target
            function. It must return true if you want the polling function to stop
            and return this value. It must return false if you want to continue
            polling. You may also use this function to collect non-success values. The
            default is a callback that tests for truthiness (anything not False, 0,
            or empty collection).
        step_function: A callback function that accepts two arguments: current_step,
            num_tries; and returns the next step value. By default, this is constant,
            but you can also pass a function that will increase or decrease the step.
            As an example, you can increase the wait time between calling the target
            function by 10 seconds every iteration until the step is 100 seconds--at
            which point it should remain constant at 100 seconds

            >>> def my_step_function(current_step: int, num_tries: int) -> int:
            >>>     return max(current_step + 10, 100)

        ignore_exceptions: You can specify a tuple of exceptions that should be caught
            and ignored on every iteration. If the target function raises one of
            these exceptions, it will be caught and the exception instance will be
            pushed to the queue of values collected during polling. Any other exceptions
            raised will be raised as normal.

    Returns:
        The first value from the target function that meets the condions of the
        check_success callback. By default, this will be the first value that is not
        None, 0, False, '', or an empty collection.
    
Poll by calling a target function until a certain condition is met. You must specify
at least a target function to be called and the step -- base wait time between
each function call.

Vendored from the [polling] package.
Poll by calling a target function until a certain condition is met . You must specify at least a target function to be called and the step -- base wait time between each function call . Vendored from the [ polling ] package .
def poll(
    target: Callable,
    step: int = 1,
    args: Optional[Sequence] = None,
    kwargs: Optional[dict] = None,
    timeout: Optional[int] = None,
    max_tries: Optional[int] = None,
    check_success: Callable = bool,
    step_function: Optional[Callable[[int, int], int]] = None,
    ignore_exceptions: Sequence = (),
):
    max_time = time.time() + timeout if timeout else None
    tries = 0
    last_item = None
    if args is None:
        args = ()
    if kwargs is None:
        kwargs = {}
    while True:
        if max_tries and tries >= max_tries:
            raise MaxCallException(last_item)
        try:
            val = target(*args, **kwargs)
            last_item = val
        except ignore_exceptions as e:
            last_item = e
        else:
            if check_success(val):
                return val
        tries += 1
        if max_time and time.time() >= max_time:
            raise TimeoutException(last_item)
        time.sleep(step)
        if step_function:
            step = step_function(step, tries)
def poll ( target : Callable , step : int = 1 , args : Optional [ Sequence ] = None , kwargs : Optional [ dict ] = None , timeout : Optional [ int ] = None , max_tries : Optional [ int ] = None , check_success : Callable = bool , step_function : Optional [ Callable [ [ int , int ] , int ] ] = None , ignore_exceptions : Sequence = ( ) , ) : max_time = time . time ( ) + timeout if timeout else None tries = 0 last_item = None if args is None : args = ( ) if kwargs is None : kwargs = { } while True : if max_tries and tries >= max_tries : raise MaxCallException ( last_item ) try : val = target ( * args , ** kwargs ) last_item = val except ignore_exceptions as e : last_item = e else : if check_success ( val ) : return val tries += 1 if max_time and time . time ( ) >= max_time : raise TimeoutException ( last_item ) time . sleep ( step ) if step_function : step = step_function ( step , tries )
Poll by calling a target function until a certain condition is met.
Poll by calling a target function until a certain condition is met .
"""
    Poll by calling a target function until a certain condition is met. You must specify
    at least a target function to be called and the step -- base wait time between
    each function call.

    Vendored from the [polling](https://github.com/justiniso/polling) package.

    Args:
        target: The target callable
        step: Step defines the amount of time to wait (in seconds)
        args: Arguments to be passed to the target function
        kwargs: Keyword arguments to be passed to the target function
        timeout: The target function will be called until the time elapsed is greater
            than the maximum timeout (in seconds). NOTE that the actual execution
            time of the function *can* exceed the time specified in the timeout. For
            instance, if the target function takes 10 seconds to execute and the timeout
            is 21 seconds, the polling function will take a total of 30 seconds (two
            iterations of the target --20s which is less than the timeout--21s,
            and a final iteration)
        max_tries: Maximum number of times the target function will be called before
            failing
        check_success: A callback function that accepts the return value of the target
            function. It must return true if you want the polling function to stop
            and return this value. It must return false if you want to continue
            polling. You may also use this function to collect non-success values. The
            default is a callback that tests for truthiness (anything not False, 0,
            or empty collection).
        step_function: A callback function that accepts two arguments: current_step,
            num_tries; and returns the next step value. By default, this is constant,
            but you can also pass a function that will increase or decrease the step.
            As an example, you can increase the wait time between calling the target
            function by 10 seconds every iteration until the step is 100 seconds--at
            which point it should remain constant at 100 seconds

            >>> def my_step_function(current_step: int, num_tries: int) -> int:
            >>>     return max(current_step + 10, 100)

        ignore_exceptions: You can specify a tuple of exceptions that should be caught
            and ignored on every iteration. If the target function raises one of
            these exceptions, it will be caught and the exception instance will be
            pushed to the queue of values collected during polling. Any other exceptions
            raised will be raised as normal.

    Returns:
        The first value from the target function that meets the condions of the
        check_success callback. By default, this will be the first value that is not
        None, 0, False, '', or an empty collection.
    """ # Condition passes, this is the only "successful" exit from the # polling function # Check the time after to make sure the poll function is called at least once
The first value from the target function that meets the condions of the
check_success callback. By default, this will be the first value that is not
None, 0, False, '', or an empty collection.
The first value from the target function that meets the condions of the check_success callback . By default this will be the first value that is not None 0 False ' ' or an empty collection .
target
Callable
The target callable
The target callable
step
int
Step defines the amount of time to wait (in seconds)
Step defines the amount of time to wait ( in seconds )
args
Optional[Sequence]
Arguments to be passed to the target function
Arguments to be passed to the target function
kwargs
Optional[dict]
Keyword arguments to be passed to the target function
Keyword arguments to be passed to the target function
timeout
Optional[int]
The target function will be called until the time elapsed is greater
than the maximum timeout (in seconds). NOTE that the actual execution
time of the function *can* exceed the time specified in the timeout. For
instance, if the target function takes 10 seconds to execute and the timeout
is 21 seconds, the polling function will take a total of 30 seconds (two
iterations of the target --20s which is less than the timeout--21s,
and a final iteration)
The target function will be called until the time elapsed is greater than the maximum timeout ( in seconds ) . NOTE that the actual execution time of the function * can * exceed the time specified in the timeout . For instance if the target function takes 10 seconds to execute and the timeout is 21 seconds the polling function will take a total of 30 seconds ( two iterations of the target -- 20s which is less than the timeout -- 21s and a final iteration )
max_tries
Optional[int]
Maximum number of times the target function will be called before
failing
Maximum number of times the target function will be called before failing
check_success
Callable
A callback function that accepts the return value of the target
function. It must return true if you want the polling function to stop
and return this value. It must return false if you want to continue
polling. You may also use this function to collect non-success values. The
default is a callback that tests for truthiness (anything not False, 0,
or empty collection).
A callback function that accepts the return value of the target function . It must return true if you want the polling function to stop and return this value . It must return false if you want to continue polling . You may also use this function to collect non - success values . The default is a callback that tests for truthiness ( anything not False 0 or empty collection ) .
step_function
Optional[Callable[[int, int], int]]
A callback function that accepts two arguments: current_step,
num_tries; and returns the next step value. By default, this is constant,
but you can also pass a function that will increase or decrease the step.
As an example, you can increase the wait time between calling the target
function by 10 seconds every iteration until the step is 100 seconds--at
which point it should remain constant at 100 seconds


A callback function that accepts two arguments : current_step num_tries ; and returns the next step value . By default this is constant but you can also pass a function that will increase or decrease the step . As an example you can increase the wait time between calling the target function by 10 seconds every iteration until the step is 100 seconds -- at which point it should remain constant at 100 seconds
ignore_exceptions
Sequence
You can specify a tuple of exceptions that should be caught
and ignored on every iteration. If the target function raises one of
these exceptions, it will be caught and the exception instance will be
pushed to the queue of values collected during polling. Any other exceptions
raised will be raised as normal.
You can specify a tuple of exceptions that should be caught and ignored on every iteration . If the target function raises one of these exceptions it will be caught and the exception instance will be pushed to the queue of values collected during polling . Any other exceptions raised will be raised as normal .