Socket
Socket
Sign inDemoInstall

waiting

Package Overview
Dependencies
0
Maintainers
1
Alerts
File Explorer

Install Socket

Protect your apps from supply chain attacks

Install

waiting

Utility for waiting for stuff to happen

    1.4.1

Maintainers
1

Readme

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 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)
 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

FAQs


Did you know?

Socket installs a GitHub app to automatically flag issues on every pull request and report the health of your dependencies. Find out what is inside your node modules and prevent malicious activity before you update the dependencies.

Install

Related posts

SocketSocket SOC 2 Logo

Product

  • Package Alerts
  • Integrations
  • Docs
  • Pricing
  • FAQ
  • Roadmap

Stay in touch

Get open source security insights delivered straight into your inbox.


  • Terms
  • Privacy
  • Security

Made with ⚡️ by Socket Inc