Metadata-Version: 2.3
Name: waiting
Version: 1.5.0
Summary: Utility for waiting for stuff to happen
Project-URL: Homepage, https://github.com/getslash/waiting
Author-email: Rotem Yaari <vmalloc@gmail.com>
License: BSD 3-Clause License
License-File: LICENSE
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.8
Provides-Extra: testing
Requires-Dist: pyforge; extra == 'testing'
Requires-Dist: pytest; extra == 'testing'
Description-Content-Type: text/x-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 <http://flux.readthedocs.org>`_ 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
