{"id":3743,"library":"polling","title":"Polling Utility","description":"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.","status":"maintenance","version":"0.3.2","language":"en","source_language":"en","source_url":"http://github.com/justiniso/polling","tags":["polling","retry","utility","wait","asynchronous"],"install":[{"cmd":"pip install polling","lang":"bash","label":"Install stable version"}],"dependencies":[],"imports":[{"note":"The primary function `poll` is usually imported directly for conciseness.","wrong":"import polling; polling.poll()","symbol":"poll","correct":"from polling import poll"},{"note":"Used for catching timeout errors and inspecting intermediate values.","symbol":"TimeoutException","correct":"from polling import TimeoutException"}],"quickstart":{"code":"import time\nfrom polling import poll, TimeoutException\n\ndef my_task_status(task_id):\n    # Simulate an external task that eventually completes\n    # In a real scenario, this would check a database, API, etc.\n    statuses = {123: ['pending', 'processing', 'completed']}\n    current_status_index = getattr(my_task_status, 'counter', 0)\n    status_list = statuses.get(task_id, ['failed'])\n    status = status_list[current_status_index % len(status_list)]\n    my_task_status.counter = current_status_index + 1\n    print(f\"Task {task_id}: current status is '{status}'\")\n    return status\n\nmy_task_status.counter = 0 # Initialize counter\n\ntry:\n    # Poll until the task status is 'completed' or timeout after 5 seconds\n    result = poll(\n        lambda: my_task_status(123),\n        check_success=lambda status: status == 'completed',\n        step=1, # Check every 1 second\n        timeout=5 # Stop after 5 seconds\n    )\n    print(f\"Task completed with status: {result}\")\nexcept TimeoutException as te:\n    print(f\"Polling timed out after {te.timeout} seconds.\")\n    print(f\"Last value before timeout: {te.values.get_nowait()}\")\n\n# Reset counter for another run (optional)\nmy_task_status.counter = 0\n\ntry:\n    # Example with ignoring exceptions (e.g., during initial setup)\n    # This dummy function will raise an error initially, then succeed\n    def flaky_check():\n        if getattr(flaky_check, 'fail_count', 0) < 2:\n            flaky_check.fail_count = getattr(flaky_check, 'fail_count', 0) + 1\n            raise ValueError(\"Still setting up...\")\n        return \"Success!\"\n    flaky_check.fail_count = 0\n\n    result_ignored = poll(\n        flaky_check,\n        ignore_exceptions=(ValueError,),\n        step=0.5,\n        timeout=3\n    )\n    print(f\"Flaky check succeeded: {result_ignored}\")\nexcept TimeoutException:\n    print(\"Flaky check timed out.\")","lang":"python","description":"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."},"warnings":[{"fix":"Migrate to `polling2` by installing `pip install polling2` and changing imports from `polling` to `polling2`.","message":"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.","severity":"gotcha","affected_versions":"0.3.0 and above"},{"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.","message":"Using `poll_forever=True` without a robust `check_success` condition or an external termination mechanism can lead to infinite loops, consuming resources indefinitely.","severity":"gotcha","affected_versions":"All"},{"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.","message":"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.","severity":"gotcha","affected_versions":"All"}],"env_vars":null,"last_verified":"2026-04-11T00:00:00.000Z","next_check":"2026-07-10T00:00:00.000Z"}