waiting 1.4.1

概述

waiting是一个等待事情发生的小库。它基本上是等待函数以各种模式返回true。

waiting与模拟时间线的flux兼容。

用法

最基本的用法是当您有一个要等待的函数时：

>>> predicate = lambda : True

永远等待很简单：

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

如果谓词返回一个值，它将作为wait（）的结果返回：

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

还可以指定timeout参数：

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

当在谓词未满的情况下超时过期时，将引发异常：

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

sleep以一定的间隔（默认为1秒）轮询谓词。可以使用sleep\u seconds参数更改间隔：

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

当等待多个谓词时，waiting提供了两个简单的工具来帮助聚合它们：any和all。它们类似于python内置的any（）和all（），只不过它们在满足谓词之后不会调用它（当谓词效率低下并且需要时间才能完成时这很有用）：

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

默认情况下，TimeoutExpired异常不会告诉您预期中没有发生的事情。要解决这个问题，请使用waiting\u for参数：

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

睡眠间隔支持指数退避：

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

如果谓词可能引发某些希望忽略的异常，则可以使用expected_exceptions忽略它们：

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

如果希望在等待谓词完成时维护更新，可以使用on_poll在每次睡眠后传递函数以执行某些行为。默认情况下，这是禁止操作。

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