Python utility library for waiting for predicates
Python
Fetching latest commit…
Cannot retrieve the latest commit at this time.
Permalink
Failed to load latest commit information.
tests
waiting
.gitignore
.travis.yml
LICENSE
MANIFEST.in
README.rst
setup.py
test_requirements.txt
tox.ini

README.rst

Overview

waiting is a small library for waiting for stuff to happen. It basically waits for a function to return True, in various modes.

Waiting is compatible with flux for simulated timelines.

Usage

The most basic usage is when you have a function you want to wait for:

>>> predicate = lambda : True

Waiting forever is very simple:

>>> from waiting import wait, TimeoutExpired
>>> wait(predicate)
True

If your predicate returns a value, it will be returned as the result of wait():

>>> result = object()
>>> wait(lambda: result) is result
True

A timeout parameter can also be specified:

>>> wait(predicate, timeout_seconds=10.5)
True

When a timeout expires without the predicate being fulfilled, an exception is thrown:

>>> try:
...     wait(lambda : False, timeout_seconds=0)
... except TimeoutExpired:
...     # expired!
...     pass
... else:
...     assert False

Sleeping polls the predicate at a certain interval (by default 1 second). The interval can be changed with the sleep_seconds argument:

>>> wait(predicate, sleep_seconds=20)
True

When waiting for multiple predicates, waiting provides two simple facilities to help aggregate them: ANY and ALL. They resemble Python's built-in any() and all(), except that they don't call a predicate once it has been satisfied (this is useful when the predicates are inefficient and take time to complete):

>>> from waiting import wait, ANY, ALL
>>> wait(ANY([predicate, predicate]))
True
>>> wait(ALL([predicate, predicate]))
True

TimeoutExpired exceptions, by default, don't tell you much about what didn't happen that you were expecting. To fix that, use the waiting_for argument:

>>> try:
...     wait(lambda : False, timeout_seconds=0, waiting_for="something that will never happen") #doctest: +ELLIPSIS
... except TimeoutExpired as e:
...     print(e)
Timeout of 0 seconds expired waiting for something that will never happen

Exponential backoff is supported for the sleep interval:

>>> from waiting import wait
>>> wait(predicate, sleep_seconds=(1, 100)) # sleep 1, 2, 4, 8, 16, 32, 64, 100, 100, ....
True
>>> wait(predicate, sleep_seconds=(1, 100, 3)) # sleep 1, 3, 9, 27, 81, 100, 100, 100 ....
True
>>> wait(predicate, sleep_seconds=(1, None)) # sleep 1, 2, 4, 6, .... (infinity)
True
>>> wait(predicate, sleep_seconds=(1, None, 4)) # sleep 1, 4, 16, 64, ... (infinity)
True

If your predicate might raise certain exceptions you wish to ignore, you may use expected_exceptions to ignore them:

>>> from waiting import wait
>>> wait(predicate, expected_exceptions=ValueError)
True
>>> wait(predicate, expected_exceptions=(ValueError, AttributeError))
True

If you'd like to maintain updates while waiting for a predicate to complete, you may use on_poll to pass a function to perform some behavior after every sleep. By default, this is a no-op.

>>> import logging
>>> from waiting import wait
>>> try:
...    wait(lambda: False, timeout_seconds=5,               # Timeout after 5 seconds
...          on_poll=lambda: logging.warn("Waiting...")) # Log "Waiting..." six times.
... except TimeoutExpired:
...    pass
... else:
...    assert False