polling2: Enhanced Polling Utility
polling2 is an updated Python polling utility that provides many configurable options for retrying operations. It allows users to repeatedly call a function until a desired condition is met, handling timeouts, delays, and error logging. The current version is 0.5.0, with new features and minor fixes released periodically.
Warnings
- breaking The library package name changed from `polling` to `polling2` in version 0.4.0. Projects using the original `polling` library must update their imports and dependencies.
- gotcha The `timeout` parameter's behavior was clarified in v0.4.7: `timeout=0` or `timeout=None` is equivalent to setting `poll_forever=True`. Misunderstanding this can lead to infinite loops if not intended.
- gotcha The `check_success` function (or equivalent lambda) must return a boolean. If it raises an exception, `polling2` will stop unless `ignore_exceptions` is configured to handle specific exceptions.
- gotcha When `poll()` fails to meet the `check_success` condition within the specified `timeout` or `max_tries`, it raises `polling2.TimeoutException` or `polling2.MaxCallException` respectively. Failing to catch these can lead to unhandled errors.
Install
-
pip install polling2
Imports
- poll
from polling2 import poll
- poll_decorator
from polling2 import poll_decorator
- TimeoutException
from polling2 import TimeoutException
- MaxCallException
from polling2 import MaxCallException
Quickstart
import time
from polling2 import poll, TimeoutException
# Simulate a resource that becomes available after some time
data_ready = False
def get_data():
global data_ready
if not data_ready:
print("Data not ready yet...")
return None
print("Data is ready!")
return {"status": "success", "data": [1, 2, 3]}
def check_data_status(result):
return result is not None and result.get("status") == "success"
# Simulate data becoming ready after 2 seconds in a separate thread
import threading
def make_data_ready_soon():
global data_ready
time.sleep(2.5)
data_ready = True
print("Attempting to poll for data...")
threading.Thread(target=make_data_ready_soon).start()
try:
# Poll for the data, checking every 0.5 seconds, with a 5-second timeout
result = poll(
target=get_data,
step=0.5,
timeout=5,
check_success=check_data_status,
# log_errors=True # Uncomment to log errors from target function
)
print(f"Polling successful: {result}")
except TimeoutException:
print("Polling timed out: Data did not become ready within 5 seconds.")
except Exception as e:
print(f"An unexpected error occurred during polling: {e}")