等待 1.5.0

概述

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

睡眠轮询会在一定的时间间隔内（默认为1秒）检查谓词。间隔可以通过 sleep_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_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.warning("Waiting...")) # Log "Waiting..." six times.
... except TimeoutExpired:
...    pass
... else:
...    assert False