Try the API, no key

# The asyncio Timeout Trap: Why wait\_for() Doesn't Stop Tasks Instantly and How to Fix It

I'll break down the common issues and offer sample code for alternatives in a friendly way!

In Python's `asyncio`, timeouts are handled primarily using the `asyncio.wait_for()` utility. It wraps a coroutine or task and raises an `asyncio.TimeoutError` if the specified duration is exceeded.

A frequent misunderstanding is that `asyncio.wait_for()` magically stops the task exactly at the timeout mark.

The Reality

```python
import asyncio

async def long_running_operation():
    print("Task started...")
    # This task *doesn't* check for cancellation or yield control
    # It will run until the end, even after timeout
    await asyncio.sleep(5) 
    print("Task finished (maybe too late!)")

async def main_problem():
    try:
        # Timeout is 1 second
        await asyncio.wait_for(long_running_operation(), timeout=1.0)
    except asyncio.TimeoutError:
        print("Timeout occurred! ") 
        # The task might still be running for a while!
    print("Main continues...")

# asyncio.run(main_problem()) 
```

\<hr>

The best practice is to structure your long-running task so that it handles `CancelledError` and performs necessary cleanup. This ensures resources are released even if a timeout occurs.

```python
import asyncio

async def gracefully_cancellable_task():
    print("Graceful task: Starting work...")
    try:
        # Simulate an operation
        await asyncio.sleep(3) 
        print("Graceful task: Work finished successfully.")
    except asyncio.CancelledError:
        #  KEY STEP: Handle cancellation for cleanup 
        print("Graceful task: Cancellation detected! Starting cleanup...")
        # Imagine closing a connection or saving state here
        await asyncio.sleep(0.1) # Cleanup time
        print("Graceful task: Cleanup finished.")
        raise # Re-raise to let the waiter know it was cancelled

async def main_graceful():
    print("Main: Starting task with 1-second timeout...")
    try:
        await asyncio.wait_for(gracefully_cancellable_task(), timeout=1.0)
    except asyncio.TimeoutError:
        print("Main: Timeout occurred! Task should be cleaning up now.")
    
    # Give the task a moment to finish cleanup before exiting (optional but safe)
    await asyncio.sleep(0.5) 
    print("Main: Program exit.")

# asyncio.run(main_graceful())
```

asyncio-compatible timeout context manager.

This library has effectively been upstreamed into Python 3.11+.

Therefore this library is considered deprecated and no longer actively supported.

Version 5.0+ provides dual-mode when executed on Python 3.11+:
`asyncio_timeout.Timeout` is fully compatible with `asyncio.Timeout` *and* old
versions of the library.

Anyway, using upstream is highly recommended. `asyncio_timeout` exists only for the
sake of backward compatibility, easy supporting both old and new Python by the same
code, and easy misgration.

If rescheduling API is not important and only `async with timeout(...): ...` functionality is required,
a user could apply conditional import:

```
if sys.version_info >= (3, 11):
    from asyncio import timeout, timeout_at
else:
    from async_timeout import timeout, timeout_at
```

The context manager is useful in cases when you want to apply timeout
logic around block of code or in cases when `asyncio.wait_for()` is
not suitable. Also it's much faster than `asyncio.wait_for()`
because `timeout` doesn't create a new task.

The `timeout(delay)` call returns a context manager
that cancels a block on *timeout* expiring:

```
from async_timeout import timeout
async with timeout(1.5):
    await inner()
```

```
async with timeout(1.5) as cm:
    # recommended api
    cm.reschedule(cm.when() + 1)  # add another second on waiting
    # compatible api
    cm.shift(1)  # add another second on waiting
    cm.update(loop.time() + 5)  # reschedule to now+5 seconds
```

Rescheduling is forbidden if the timeout is expired or after exit from `async with`
code block.

Disable scheduled timeout:

```
async with timeout(1.5) as cm:
    cm.reschedule(None)  # recommended api
    cm.reject()          # compatible api
```

```
$ pip install async-timeout
```

The library is Python 3 only!

The module is written by Andrew Svetlov.

It's *Apache 2* licensed and freely available.

# The async with timeout: Best Practices for Robust Python Asynchronous I/O

Here's a friendly breakdown of the `asyncio.timeout()` context manager, common pitfalls, and reliable alternatives.

The `asyncio.timeout()` function provides a way to impose a time limit on an `awaitable` operation (like an asynchronous function, a task, or an I/O operation) within a specific block of code.

It works as an `async` context manager

When you enter the `async with asyncio.timeout(delay):` block, a timer starts.

If the code inside the block finishes before the time limit (`delay`) expires, everything is normal.

If the time limit is reached while the inner code is suspended at an `await` point, the execution is cancelled, and an `asyncio.TimeoutError` is raised at that `await` point.

This example shows how to use it to limit the execution time of an operation.

```python
import asyncio

async def long_running_operation():
    print("Operation started...")
    await asyncio.sleep(5)  # Simulating a long I/O wait
    print("Operation finished successfully!")
    return "Data Acquired"

async def main():
    try:
        # Set a timeout of 3 seconds
        async with asyncio.timeout(3) as cm:
            result = await long_running_operation()
            print(f"Result: {result}")
    except asyncio.TimeoutError:
        print(" Operation timed out!")
        # You can access details about the remaining time via the context manager
        # print(f"Time remaining at timeout: {cm.when()}")

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

# Expected Output (since 3 < 5):
# Operation started...
#  Operation timed out!
```

```python
import time

async def blocking_code_issue():
    async with asyncio.timeout(1):
        #  This is synchronous and blocks the entire event loop for 5 seconds.
        # The timeout cannot interrupt it.
        print("Starting blocking work...")
        time.sleep(5) 
        print("Finished blocking work (timeout was ineffective)") 
        # The TimeoutError will only raise *after* this finishes, 
        # or at the next await, which is often too late or confusing.
```

To handle long-running, CPU-bound (blocking) operations, you should run them in a separate thread or process pool using `asyncio.to_thread()` (Python 3.9+) or `loop.run_in_executor()`.

**Summary**: in this tutorial, you’ll learn how to use the `asyncio.wait_for()` function to wait for a coroutine to complete with a timeout.

## Introduction to the Python asyncio.wait\_for() function #

In the previous tutorial, you learned how to cancel a task that is in progress by using the `cancel()` method of the `Task` object.

To wait for a task to complete with a timeout, you can use the `asyncio.wait_for()` function. The `asyncio.wait_for()` function waits for a single task to be completed with a timeout.

When a timeout occurs, the `asyncio.wait_for()` function cancels the task and raises the `TimeoutError` exception. Otherwise, it returns the result of the task. For example:

```python
import asyncio
from asyncio.exceptions import TimeoutError


async def call_api(message, result=1000, delay=3):
    print(message)
    await asyncio.sleep(delay)
    return result


async def main():
    task = asyncio.create_task(
        call_api('Calling API...', result=2000, delay=5)
    )

    MAX_TIMEOUT = 3
    try:
        await asyncio.wait_for(task, timeout=MAX_TIMEOUT)
    except TimeoutError:
        print('The task was cancelled due to a timeout')

asyncio.run(main())
Copy
```

Output:

```abap
Calling API...
The task was cancelled due to a timeoutCopy
```

How it works

First, define a `call_api()` coroutine that takes 3 seconds to complete by default:

```python
async def call_api(message, result=1000, delay=3):
    print(message)
    await asyncio.sleep(delay)
    return resultCopy
```

Second, create a task that wraps the call\_api coroutine and takes 5 seconds to complete:

```nginx
task = asyncio.create_task(
    call_api('Calling API...', result=2000, delay=5)
)Copy
```

Third, use the `asyncio.wait_for()` function to wait for the task to complete with a timeout of 3 seconds. Since the task takes 5 seconds to complete, a timeout will occur and a `TimeoutError` will be raised:

# High-level API Index

This page lists all high-level async/await enabled asyncio APIs.

## Tasks

Utilities to run asyncio programs, create Tasks, and await on multiple things with timeouts.

| `run()` | Create event loop, run a coroutine, close the loop. | 
| `Runner` | A context manager that simplifies multiple async function calls. | 
| `Task` | Task object. | 
| `TaskGroup` | A context manager that holds a group of tasks. Provides a convenient and reliable way to wait for all tasks in the group to finish. | 
| `create_task()` | Start an asyncio Task, then returns it. | 
| `current_task()` | Return the current Task.