{"id":8770,"library":"wait-for2","title":"wait-for2 Asyncio Timeout Utility","description":"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.","status":"active","version":"0.4.1","language":"en","source_language":"en","source_url":"https://github.com/Traktormaster/wait-for2","tags":["asyncio","concurrency","timeout","cancellation","python"],"install":[{"cmd":"pip install wait-for2","lang":"bash","label":"Install wait-for2"}],"dependencies":[],"imports":[{"note":"The primary function is named 'wait_for2' to distinguish it from 'asyncio.wait_for'. Ensure you import the function, not just the package.","wrong":"from wait_for2 import wait_for","symbol":"wait_for2","correct":"from wait_for2 import wait_for2"}],"quickstart":{"code":"import asyncio\nfrom wait_for2 import wait_for2\n\nasync def task_that_might_timeout():\n    \"\"\"A mock async task that takes some time to complete.\"\"\"\n    await asyncio.sleep(0.6)\n    return \"Task completed!\"\n\nasync def main():\n    print(\"--- Basic Timeout Example ---\")\n    try:\n        # Attempt to run the task with a timeout shorter than its execution time\n        result = await wait_for2(task_that_might_timeout(), timeout=0.5)\n        print(f\"Success: {result}\")\n    except asyncio.TimeoutError:\n        print(\"Timeout: Task did not complete in time.\")\n\n    print(\"\\n--- Race Handler Example (v0.3.0+ signature) ---\")\n    async def fast_task():\n        await asyncio.sleep(0.1)\n        return \"Fast result\"\n\n    async def slow_task():\n        await asyncio.sleep(1.0)\n        return \"Slow result\"\n\n    # Define a custom race handler that inspects the result and exception status\n    async def custom_race_handler(result, is_exception):\n        if is_exception:\n            print(f\"  Handler caught exception: {result.__class__.__name__}\")\n            return \"Handled exception scenario\"\n        print(f\"  Handler caught completion: {result}\")\n        return f\"Handled: {result}\"\n\n    # Scenario 1: Task completes within timeout, handler gets completion result\n    result_complete = await wait_for2(fast_task(), timeout=0.5, race_handler=custom_race_handler)\n    print(f\"  wait_for2 result (completion): {result_complete}\")\n\n    # Scenario 2: Task times out, handler gets TimeoutError\n    result_timeout = await wait_for2(slow_task(), timeout=0.2, race_handler=custom_race_handler)\n    print(f\"  wait_for2 result (timeout): {result_timeout}\")\n\nif __name__ == \"__main__\":\n    asyncio.run(main())\n","lang":"python","description":"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."},"warnings":[{"fix":"Update your `race_handler` function to accept two arguments: `async def my_handler(result, is_exception): ...`","message":"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.","severity":"breaking","affected_versions":">=0.3.0"},{"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.","message":"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.","severity":"gotcha","affected_versions":">=0.4.0 (Python 3.12+)"},{"fix":"Wrap calls to `wait_for2` in a `try...except asyncio.TimeoutError` block or provide a `race_handler` callback to manage timeout outcomes.","message":"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`.","severity":"gotcha","affected_versions":"All versions"}],"env_vars":null,"last_verified":"2026-04-16T00:00:00.000Z","next_check":"2026-07-15T00:00:00.000Z","problems":[{"fix":"Update your `race_handler` definition to accept `result` and `is_exception`: `async def my_handler(result, is_exception): ...`","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`).","error":"TypeError: race_handler() takes 1 positional argument but 2 were given"},{"fix":"Correct your import statement to `from wait_for2 import wait_for2`.","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.","error":"AttributeError: module 'wait_for2' has no attribute 'wait_for'"},{"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.","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.","error":"asyncio.TimeoutError: Operation timed out"}]}