The `polling` library provides a simple yet powerful utility to repeatedly call a function until a desired condition is met or a timeout occurs. It is useful for waiting on external resources, API responses, or file system changes. The latest version is 0.3.2, released in May 2021. While functional, it is considered to be in maintenance mode, with a more actively developed and recommended fork, `polling2`, available.
pip install pollingVerified import paths — ran on the pinned version, not inferred.
This quickstart demonstrates how to use `polling.poll` to wait for a simulated asynchronous task to complete. It shows how to define a success condition using `check_success`, set polling intervals with `step`, and handle timeouts with `TimeoutException`. An additional example illustrates how to `ignore_exceptions` during polling.
Migrate to `polling2` by installing `pip install polling2` and changing imports from `polling` to `polling2`.
Always define a clear `timeout` parameter or ensure that your `check_success` lambda/function will eventually return a truthy value or that an external event can terminate the polling process.
Catch `TimeoutException` and inspect `te.values` to debug why the condition was not met. Use `te.values.get_nowait()` or iterate the queue to access the collected values.
Increase the `timeout` or `max_tries` parameters, or ensure the `target` function is expected to succeed within the given limits. Handle the exception gracefully using a `try-except` block.
```python
import polling
import time
def my_task():
# Simulate a task that sometimes succeeds
if time.time() % 5 < 1: # Succeeds for 1 second every 5 seconds
return True
return False
try:
polling.poll(lambda: my_task(), step=0.1, timeout=2)
print("Task succeeded!")
except polling.TimeoutException as e:
print(f"Polling timed out: {e}")
# Access collected values, if any, from e.values
# while not e.values.empty():
# print(e.values.get())
```Always provide at least the `target` function and the `step` interval to `polling.poll()`.
```python
import polling
import time
def check_status():
# Your condition check here
return True
# Correct usage:
polling.poll(target=check_status, step=1, timeout=10)
# Incorrect usage (leads to error):
# polling.poll()
# polling.poll(target=check_status)
# polling.poll(step=1)
```If you intend to access the values collected during the polling attempts, access the `values` attribute of the `TimeoutException` object, which is a `queue.Queue` (or `collections.deque` in some versions/forks) and is iterable. Ensure `collect_values=True` was passed to `polling.poll()` if you want values to be collected.
```python
import polling
import random
try:
polling.poll(lambda: random.choice([0, (), False]), step=0.5, timeout=1, collect_values=True)
except polling.TimeoutException as te:
print("Polling timed out. Values that did not meet the condition:")
# Correct way to access collected values
while not te.values.empty():
print(te.values.get())
# Incorrect way (leads to TypeError):
# except polling.TimeoutException as te:
# for value in te: # This would cause the error
# print(value)
```No dependency data recorded yet.