Easy and expressive busy-waiting for Python

0.6.1 · active · verified Fri Apr 17

Busypie is a Python library for expressive busy-waiting, primarily used in integration and end-to-end tests to await conditions. It provides a fluent API for defining timeouts, retry intervals, and custom error messages. The current version is 0.6.1, with a release cadence that has seen several updates in the past year.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates basic busy-waiting using `wait().until()` and handling `ConditionTimeoutError`. It also shows the usage of `at_most()` for maximum wait time, `poll_interval()` for checking frequency, and `at_least()` for specifying a minimum duration a condition must be met for (introduced in v0.6.0).

from busypie import wait, ConditionTimeoutError
import time

# Simulate an external service state
_external_service_status = 'starting'

def get_service_status():
    global _external_service_status
    if _external_service_status == 'starting':
        # Simulate service becoming ready after some time
        time.sleep(0.05) 
        _external_service_status = 'ready'
    return _external_service_status

def is_service_ready():
    return get_service_status() == 'ready'

def process_completes_slowly():
    time.sleep(0.4)
    return True

try:
    print('Waiting for service to be ready...')
    wait().at_most(1).poll_interval(0.1).until(is_service_ready)
    print('Service is ready!')

    # Example with at_least (v0.6.0+)
    print('Waiting for a process that must take at least 0.3s...')
    wait().at_most(1).at_least(0.3).until(process_completes_slowly)
    print('Process completed and met minimum duration.')

except ConditionTimeoutError as e:
    print(f'A condition timed out: {e}')

_external_service_status = 'starting'
try:
    # This will timeout because the condition is met too quickly, 
    # but 'at_least' expects a longer duration.
    print('Waiting for a process that must take at least 1s (will fail)...')
    wait().at_most(0.5).at_least(1).until(is_service_ready)
    print('Process completed and met minimum duration (unexpected).')
except ConditionTimeoutError as e:
    print(f'Caught expected timeout for at_least condition: {e}')

view raw JSON →