Polling Utility

0.3.2 · maintenance · verified Sat Apr 11

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.

Warnings

gotcha The `polling` library (justiniso/polling) is no longer actively maintained. For ongoing development and better support, consider using the `polling2` library (ddmee/polling2), which is a direct fork and actively developed.
Fix: Migrate to `polling2` by installing `pip install polling2` and changing imports from `polling` to `polling2`.
gotcha Using `poll_forever=True` without a robust `check_success` condition or an external termination mechanism can lead to infinite loops, consuming resources indefinitely.
Fix: 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.
gotcha When `polling.poll` times out, it raises a `TimeoutException`. This exception includes a `values` attribute (a queue) containing all values returned by the target function that did not meet the `check_success` condition before the timeout.
Fix: 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.

Install

pip install polling Install stable version

Imports

poll
```
from polling import poll
```
The primary function `poll` is usually imported directly for conciseness.
TimeoutException
```
from polling import TimeoutException
```
Used for catching timeout errors and inspecting intermediate values.

Quickstart

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.

import time
from polling import poll, TimeoutException

def my_task_status(task_id):
    # Simulate an external task that eventually completes
    # In a real scenario, this would check a database, API, etc.
    statuses = {123: ['pending', 'processing', 'completed']}
    current_status_index = getattr(my_task_status, 'counter', 0)
    status_list = statuses.get(task_id, ['failed'])
    status = status_list[current_status_index % len(status_list)]
    my_task_status.counter = current_status_index + 1
    print(f"Task {task_id}: current status is '{status}'")
    return status

my_task_status.counter = 0 # Initialize counter

try:
    # Poll until the task status is 'completed' or timeout after 5 seconds
    result = poll(
        lambda: my_task_status(123),
        check_success=lambda status: status == 'completed',
        step=1, # Check every 1 second
        timeout=5 # Stop after 5 seconds
    )
    print(f"Task completed with status: {result}")
except TimeoutException as te:
    print(f"Polling timed out after {te.timeout} seconds.")
    print(f"Last value before timeout: {te.values.get_nowait()}")

# Reset counter for another run (optional)
my_task_status.counter = 0

try:
    # Example with ignoring exceptions (e.g., during initial setup)
    # This dummy function will raise an error initially, then succeed
    def flaky_check():
        if getattr(flaky_check, 'fail_count', 0) < 2:
            flaky_check.fail_count = getattr(flaky_check, 'fail_count', 0) + 1
            raise ValueError("Still setting up...")
        return "Success!"
    flaky_check.fail_count = 0

    result_ignored = poll(
        flaky_check,
        ignore_exceptions=(ValueError,),
        step=0.5,
        timeout=3
    )
    print(f"Flaky check succeeded: {result_ignored}")
except TimeoutException:
    print("Flaky check timed out.")

view raw JSON →