Examples¶

Poll under the current time in seconds is divisble by 5¶

(Ignoring milliseconds)

import polling2, time
# Wait until the number of seconds (ignoring milliseconds) is divisible by 5.
polling2.poll(target=time.time, check_success=lambda x: int(x) % 5 == 0, step=0.5, timeout=6)
1599737060.4507122

Poll every minute until a url returns 200 status code¶

import requests
polling2.poll(
    lambda: requests.get('http://google.com').status_code == 200,
    step=60,
    poll_forever=True)

Poll but ignore specifc exceptions¶

If you are creating a new cloud provider instance (e.g. waiting for an EC2 instance to come online), you can continue to poll despite getting ConnectionErrors

import requests
polling2.poll(
    lambda: requests.get('your.instance.ip').status_code == 200,
    step=60,
    ignore_exceptions=(requests.exceptions.ConnectionError,),
    poll_forever=True)

Poll for a file to exist¶

This call will wait until the file exists, checking every 0.1 seconds and stopping after 3 seconds have elapsed

file_handle = polling2.poll(lambda: open('/tmp/myfile.txt'), ignore_exceptions=(IOError,), timeout=3, step=0.1)
# Polling will return the value of your polling function, so you can now interact with it
file_handle.close()

Note, poll returns the value of whatever the target last returned.

Polling for Selenium WebDriver elements¶

from selenium import webdriver

driver = webdriver.Firefox()
driver.get('http://google.com')

search_box = polling2.poll(lambda: driver.find_element_by_id('search'), step=0.5, timeout=7)
search_box.send_keys('python polling')

Using the polling timeout exception¶

An exception will be raised by the polling function on timeout (or the maximum number of calls is exceeded). This exception will have a ‘values’ attribute. This is a queue with all values that did not meet the condition. You can access them in the except block.

import random

try:
    polling2.poll(lambda: random.choice([0, (), False]), step=0.5, timeout=1)
except polling2.TimeoutException, te:
    while not te.values.empty():
        # Print all of the values that did not meet the exception
        print te.values.get()

Use a custom checker to decide whether your target has returned what your waiting for¶

is_truthy() is the default checker for the parameter check_success. But, it’s easy to create a custom checker function, that tests whether the value returned by the target is the expected value.

Here the target is going to return None, which the custom checker, created by is_value(None) will return True for.

polling2.poll(target=lambda: None, step=0.1, max_tries=1, check_success=polling2.is_value(None))
# Or another example, where we can test that False is returned by the target.
polling2.poll(target=lambda: False, step=0.1, max_tries=1, check_success=polling2.is_value(False))

Using a custom condition callback function¶

import requests

def is_correct_response(response):
    """Check that the response returned 'success'"""
    return response == 'success'

polling2.poll(
    lambda: requests.put('http://mysite.com/api/user', data={'username': 'Jill'},
    check_success=is_correct_response,
    step=1,
    timeout=10)

Logging the return values from the target function¶

import logging
import requests

def is_correct_response(response):
    """Check that the response returned 'success'"""
    return response == 'success'

polling2.poll(
    lambda: requests.put('http://mysite.com/api/user', data={'username': 'Jill'},
    check_success=is_correct_response,
    step=1,
    timeout=10,
    log=logging.DEBUG)

This will log the string representation of response object to python’s logging module at the debug level. A message like this will be sent to the log for each return value. You can change the level by providing a different value to the log parameter.

poll() calls check_success(<Response [200]>)

There is also an option to log the exceptions that are caught by ignore_exceptions. Note, the full-exception traceback will not be printed in the logs. Instead, the error and it’s message (using %r formatting) will appear. In the following code snippet, the ValueError raised by the function raises_error() will be sent to the logger at the ‘warning’ level.

import polling2
import logging
import mock

# basicConfig should sent warning level messages to the stdout.
logging.basicConfig()

# Create a function that raises a ValueError, then a RuntimeError.
raises_error = mock.Mock(side_effect=[ValueError('a message'), RuntimeError])

try:
    polling2.poll(
        target=raises_error,
        step=0.1,
        max_tries=3,
        ignore_exceptions=(ValueError),  # Only ignore the ValueError.
        log_error=logging.WARNING  # Ignored errors should be passed to the logger at warning level.
    )
except RuntimeError as _e:
    print "Un-ignored %r" % _e"

Poll a target forever¶

If you do not want to set a timeout on the polled target, you can set the poll_forever parameter to true. This will poll the target forever until the target returns the value you expect.

from polling2 import poll
# Target function returns False. The default check_success looks for a truthy value to be returned.
# This will call the target function forever, with a one second wait between polls.
poll(target=lambda: False, step=1, poll_forever=True)

Note, that setting the timeout parameter to None or 0 has the equivalent effect as setting the poll_forever parameter to true. E.g.:

from polling2 import poll
# Setting timeout to zero is equivalent to setting poll_forever=True.
# This call will also poll the target forever.
poll(target=lambda: False, step=1, timeout=0)

Wrap a target function in a polling decorator¶

Perhaps you’d like to use the decorator syntax to implement the polling functional. No problem!

from polling2 import poll_decorator
import requests
@poll_decorator(step=1, timeout=15)
def wait_until_exists(uri):
    return requests.get(uri).status_code != 404
# Call when you please
wait_until_exists(uri='http://www.google.com')

Inspiration taken from https://github.com/benjamin-hodgson/poll per request from https://github.com/lucasmelin.