Skip to main content

Utility for waiting for stuff to happen

Project description


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.


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)

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

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

A timeout parameter can also be specified:

>>> wait(predicate, timeout_seconds=10.5)

When a timeout expires without the predicate being fullfilled, 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)

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]))
>>> wait(ALL([predicate, predicate]))

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, ....
>>> wait(predicate, sleep_seconds=(1, 100, 3)) # sleep 1, 3, 9, 27, 81, 100, 100, 100 ....
>>> wait(predicate, sleep_seconds=(1, None)) # sleep 1, 2, 4, 6, .... (infinity)
>>> wait(predicate, sleep_seconds=(1, None, 4)) # sleep 1, 4, 16, 64, ... (infinity)

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)
>>> wait(predicate, expected_exceptions=(ValueError, AttributeError))

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

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

waiting-1.4.1.tar.gz (7.1 kB view hashes)

Uploaded Source

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page