Waiter Library for Polling and Retries

Common errors

• TypeError: 'function' object is not iterable

  cause You are passing a callable function to `waiter.wait`, which expects an iterable (like `range(N)` or a list).
  fix Provide an iterable object to `waiter.wait`. If you intended to poll a function for success, use `waiter.until` or `waiter.until_successful`. Correct example: `for _ in waiter.wait(range(5), pause=1): print('Waiting...')`

• TypeError: '<callable_name>' object is not callable

  cause You are passing the *result* of a function call to `waiter.until` or `waiter.until_successful` instead of the function object itself, or passing a non-callable object.
  fix Ensure you pass the function reference (the function name without parentheses) to `until` or `until_successful`. Correct example: `until_successful(my_function, timeout=5)` instead of `until_successful(my_function(), timeout=5)`.

• TimeoutError: Callable did not return a truthy value or stop raising exceptions within X seconds

  cause The target callable for `waiter.until` or `waiter.until_successful` did not meet its success condition (returned a truthy value/stopped raising exceptions) within the `timeout` period.
  fix Increase the `timeout` parameter if the operation genuinely takes longer, or decrease `pause` if more frequent checks are needed. Review the logic of your callable to ensure it eventually succeeds. Always catch `TimeoutError` to handle the failure gracefully.

Warnings

• breaking The `sleep_seconds` argument in `waiter.wait`, `until`, and `until_successful` was renamed to `pause` in version 1.3.0.
  Fix: Update your code to use `pause` instead of `sleep_seconds`. For example, `waiter.until(my_func, pause=1)`.
• gotcha `waiter.wait` expects an *iterable* (e.g., `range`, a list, or a generator) as its first argument, not a callable function. Passing a function will result in a `TypeError`.
  Fix: Ensure you provide an iterable object to `waiter.wait`. If you intend to poll a function, use `waiter.until` or `waiter.until_successful` instead. Example: `for _ in waiter.wait(range(10), pause=0.5): ...`
• gotcha `waiter.until_successful` only retries when the target callable *raises an exception*. It does *not* retry if the callable returns a falsy value (e.g., `None` or `False`). For that, you should use `waiter.until`.
  Fix: If your condition for retrying is a falsy return value, use `waiter.until`. If it's an exception, use `waiter.until_successful`. For example, `until(check_condition, timeout=10)` will retry if `check_condition()` returns `False` or `None`.
• gotcha If the polling condition is not met within the specified `timeout`, `waiter` functions will raise a `TimeoutError`. Forgetting to handle this can lead to unhandled exceptions.
  Fix: Always wrap your `waiter` calls in a `try...except TimeoutError` block to gracefully handle scenarios where the condition isn't met in time.

Install

• pip install waiter Install stable version

Imports

• wait

  from waiter import wait

  Use for delayed iteration over an iterable.
• until

  from waiter import until

  Use for polling until a callable returns a truthy value.
• until_successful

  from waiter import until_successful

  Use for retrying a callable until it stops raising exceptions.

Quickstart

This quickstart demonstrates using `waiter.until_successful` to poll a function that initially fails but eventually succeeds. It illustrates how to define a retriable operation and handle potential `TimeoutError`.

import time
from waiter import until_successful

# Define a function that simulates checking a service or condition
class ServiceChecker:
    def __init__(self):
        self.attempts = 0

    def check_status(self):
        self.attempts += 1
        if self.attempts < 3:
            # Simulate a temporary failure by raising an exception
            print(f"Attempt {self.attempts}: Service not ready yet...")
            raise ConnectionError("Service not available")
        else:
            # Simulate success after a few attempts
            print(f"Attempt {self.attempts}: Service is ready!")
            return f"Service ready after {self.attempts} attempts"

checker = ServiceChecker()
result = None
try:
    # Use until_successful to retry calling the check_status method.
    # It will keep retrying until the callable succeeds (doesn't raise an exception)
    # or the timeout is reached.
    result = until_successful(
        checker.check_status,
        timeout=5,  # Max time to wait in seconds
        pause=0.5   # Time to wait between retries in seconds
    )
    print(f"Polling successful: {result}")
except TimeoutError as e:
    print(f"Polling failed: {e}")
except Exception as e:
    print(f"An unexpected error occurred: {e}")