Skip to main content

Utility for waiting for stuff to happen

Project description

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.warning("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.5.0.tar.gz (6.9 kB view hashes)

Uploaded Source

Built Distribution

waiting-1.5.0-py3-none-any.whl (5.7 kB view hashes)

Uploaded Python 3

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