wait-for2 Asyncio Timeout Utility

0.4.1 · active · verified Thu Apr 16

wait-for2 provides an enhanced `asyncio.wait_for` function designed to robustly handle complex scenarios involving simultaneous task completion and cancellation. It aims to prevent race conditions where a task might complete just as it's being cancelled. The current version is 0.4.1. Releases are made on an as-needed basis, typically for bug fixes, Python version compatibility, or new features.

Common errors

```
TypeError: race_handler() takes 1 positional argument but 2 were given
```
cause Your `race_handler` callback was defined with only one argument, but `wait-for2` v0.3.0 and later pass two arguments (`result`, `is_exception`).

fix Update your `race_handler` definition to accept `result` and `is_exception`: `async def my_handler(result, is_exception): ...`
```
AttributeError: module 'wait_for2' has no attribute 'wait_for'
```
cause You are trying to import `wait_for` from the `wait_for2` package, but the main function is explicitly named `wait_for2` to differentiate it.

fix Correct your import statement to `from wait_for2 import wait_for2`.
```
asyncio.TimeoutError: Operation timed out
```
cause Your coroutine exceeded the specified `timeout` duration, and you did not provide a `race_handler` to gracefully manage this outcome, nor did you wrap the `wait_for2` call in a `try...except asyncio.TimeoutError` block.

fix Either add a `try...except asyncio.TimeoutError` block around your `await wait_for2(...)` call, or provide a `race_handler` callback (e.g., `await wait_for2(task(), timeout=1, race_handler=my_handler)`) to handle the timeout internally.

Warnings

breaking The `race_handler` callback signature changed in v0.3.0. It now receives two arguments: `(result, is_exception)`, where `is_exception` is a boolean indicating if the `result` is an exception.
Fix: Update your `race_handler` function to accept two arguments: `async def my_handler(result, is_exception): ...`
gotcha For Python 3.12 and newer, `wait-for2`'s implementation internally prefers the builtin `asyncio.wait_for` where possible. While the API remains consistent, subtle behavioral differences might arise in extremely complex cancellation scenarios due to changes in the underlying asyncio primitives.
Fix: Generally no fix required as API compatibility is maintained. If encountering unexpected behavior in advanced cancellation logic, test carefully or consider previous Python versions for comparison if 3.12+ specific.
gotcha If you don't provide a `race_handler`, `wait_for2` will raise an `asyncio.TimeoutError` on timeout. This needs to be explicitly handled with a `try...except asyncio.TimeoutError` block, similar to the standard `asyncio.wait_for`.
Fix: Wrap calls to `wait_for2` in a `try...except asyncio.TimeoutError` block or provide a `race_handler` callback to manage timeout outcomes.

Install

pip install wait-for2 Install wait-for2

Imports

wait_for2
wrong:
```
from wait_for2 import wait_for
```
correct:
```
from wait_for2 import wait_for2
```
The primary function is named 'wait_for2' to distinguish it from 'asyncio.wait_for'. Ensure you import the function, not just the package.

Quickstart

This quickstart demonstrates the basic usage of `wait_for2` with a timeout, and then showcases the `race_handler` callback, including its 2-argument signature introduced in v0.3.0. The `race_handler` allows custom logic for outcomes where a task either completes or times out/gets cancelled.

import asyncio
from wait_for2 import wait_for2

async def task_that_might_timeout():
    """A mock async task that takes some time to complete."""
    await asyncio.sleep(0.6)
    return "Task completed!"

async def main():
    print("--- Basic Timeout Example ---")
    try:
        # Attempt to run the task with a timeout shorter than its execution time
        result = await wait_for2(task_that_might_timeout(), timeout=0.5)
        print(f"Success: {result}")
    except asyncio.TimeoutError:
        print("Timeout: Task did not complete in time.")

    print("\n--- Race Handler Example (v0.3.0+ signature) ---")
    async def fast_task():
        await asyncio.sleep(0.1)
        return "Fast result"

    async def slow_task():
        await asyncio.sleep(1.0)
        return "Slow result"

    # Define a custom race handler that inspects the result and exception status
    async def custom_race_handler(result, is_exception):
        if is_exception:
            print(f"  Handler caught exception: {result.__class__.__name__}")
            return "Handled exception scenario"
        print(f"  Handler caught completion: {result}")
        return f"Handled: {result}"

    # Scenario 1: Task completes within timeout, handler gets completion result
    result_complete = await wait_for2(fast_task(), timeout=0.5, race_handler=custom_race_handler)
    print(f"  wait_for2 result (completion): {result_complete}")

    # Scenario 2: Task times out, handler gets TimeoutError
    result_timeout = await wait_for2(slow_task(), timeout=0.2, race_handler=custom_race_handler)
    print(f"  wait_for2 result (timeout): {result_timeout}")

if __name__ == "__main__":
    asyncio.run(main())

view raw JSON →