Of course! Handling timeouts is a crucial aspect of writing robust and resilient asyncio applications. Here’s a comprehensive guide to using timeouts in Python's asyncio, covering the modern, recommended approach and the older, deprecated methods.

The Modern & Recommended Approach: asyncio.timeout()

Since Python 3.11, the asyncio library has a built-in, context-manager-based way to handle timeouts. This is the preferred and most robust method.

Key Concept: asyncio.timeout()

You wrap your code block in an async with asyncio.timeout(...): statement. If the block doesn't complete within the specified timeout, it raises an asyncio.TimeoutError.

Example 1: Basic Usage

Let's simulate a long-running task and see how the timeout interrupts it.

import asyncio
import time
async def long_running_task():
    print("Task: Starting a 5-second task...")
    await asyncio.sleep(5)  # Simulate work
    print("Task: Task finished successfully!")
    return "Done!"
async def main():
    print("Main: Starting with a 2-second timeout.")
    try:
        # This will timeout because the task takes 5s, but we only allow 2s
        async with asyncio.timeout(2):
            result = await long_running_task()
            print(f"Main: Got result: {result}")
    except asyncio.TimeoutError:
        print("Main: Caught the timeout! The task was cancelled.")
    print("Main: Continuing with other work...")
asyncio.run(main())

Output:

Main: Starting with a 2-second timeout.
Task: Starting a 5-second task...
Main: Caught the timeout! The task was cancelled.
Main: Continuing with other work...

How it works:

1. asyncio.timeout(2) starts a 2-second countdown.
2. await long_running_task() starts. It will try to sleep for 5 seconds.
3. After 2 seconds, the timeout expires.
4. asyncio cancels the long_running_task. This is key: it doesn't just wait; it actively stops the task.
5. The async with block raises an asyncio.TimeoutError.
6. The try...except block catches this error, and your program can handle it gracefully.

Example 2: Handling Cancellation in a Task

It's good practice for a task to handle its own cancellation. When a task is cancelled, an asyncio.CancelledError is raised within it. You can catch this to perform cleanup.

import asyncio
async def task_with_cleanup():
    print("Task: Starting task with cleanup.")
    try:
        # This is where the actual work would be
        await asyncio.sleep(10) 
    except asyncio.CancelledError:
        print("Task: I was cancelled! Performing cleanup...")
        # Simulate cleanup work
        await asyncio.sleep(1)
        print("Task: Cleanup complete.")
        # Important: Re-raise the exception to signal cancellation correctly
        raise
    print("Task: Task finished successfully (this line won't be reached).")
    return "All clean!"
async def main():
    print("Main: Starting with a 3-second timeout.")
    try:
        async with asyncio.timeout(3):
            await task_with_cleanup()
    except asyncio.TimeoutError:
        print("Main: Timeout caught. The task should have cleaned up.")
    print("Main: Waiting a bit to see the cleanup...")
    await asyncio.sleep(2) # Give the cancelled task a moment to run its cleanup
    print("Main: All done.")
asyncio.run(main())

Output:

Main: Starting with a 3-second timeout.
Task: Starting task with cleanup.
Main: Timeout caught. The task should have cleaned up.
Main: Waiting a bit to see the cleanup...
Task: I was cancelled! Performing cleanup...
Task: Cleanup complete.
Main: All done.

Notice how the task gracefully handled its own cancellation before the main program continued.

The Older, Deprecated Methods (For Context)

If you are using Python 3.10 or older, you don't have asyncio.timeout(). You would have used one of these patterns. It's recommended to upgrade to Python 3.11+ if possible, but understanding these is useful for maintaining older code.

Method 1: asyncio.wait_for()

This is a simple function that takes a coroutine and a timeout value. It waits for the coroutine to finish, but if it exceeds the timeout, it cancels the coroutine and raises asyncio.TimeoutError.

# Python 3.10 and older
import asyncio
async def my_slow_coro():
    await asyncio.sleep(3)
    return "Finished!"
async def main_old():
    print("Using wait_for with a 2-second timeout...")
    try:
        # wait_for will run my_slow_coro but cancel it after 2 seconds
        result = await asyncio.wait_for(my_slow_coro(), timeout=2.0)
        print(f"Result: {result}")
    except asyncio.TimeoutError:
        print("wait_for timed out!")
asyncio.run(main_old())

Output:

Using wait_for with a 2-second timeout...
wait_for timed out!

Caveat: wait_for can be tricky. If the coroutine you're waiting for doesn't respond to cancellation (e.g., it's in a tight loop calling a synchronous, non-cancellable function), wait_for will wait until the coroutine finishes, even if it's past the timeout.

Method 2: Manual Timeout with asyncio.shield() and asyncio.wait()

This is the most flexible and "low-level" approach. It gives you explicit control over what gets cancelled.

The pattern is:

1. asyncio.shield(): Protects a task from being cancelled.
2. asyncio.wait(): Wait for either the protected task to complete OR for the timeout to occur.

# Python 3.10 and older
import asyncio
async def protected_coro():
    print("Protected coro: Running...")
    await asyncio.sleep(3)
    print("Protected coro: Finished!")
    return "Protected data"
async def main_manual():
    print("Manual timeout method...")
    protected_task = asyncio.create_task(protected_coro())
    # We wait for EITHER the protected_task to complete OR for a timeout signal
    done, pending = await asyncio.wait(
        [protected_task],
        timeout=2.0,
        return_when=asyncio.FIRST_COMPLETED
    )
    if pending:
        # The timeout occurred, so the protected_task is still pending
        print("Manual timeout: Cancelling the protected task.")
        for p_task in pending:
            p_task.cancel() # Explicitly cancel the pending task
        # Wait for the cancelled task to finish its cancellation handling
        await asyncio.wait(pending)
        print("Manual timeout: Task was cancelled.")
    else:
        # The task finished before the timeout
        print(f"Manual timeout: Task finished with result: {protected_task.result()}")
asyncio.run(main_manual())

Output:

Manual timeout method...
Protected coro: Running...
Manual timeout: Cancelling the protected task.
Protected coro: Finished!  # Note: The task still runs to completion because it was shielded
Manual timeout: Task was cancelled.

This method is complex and shows why asyncio.timeout() is such a welcome addition—it handles all this shielding and cancellation logic for you.

Summary: Which one should I use?

Final Recommendation:

Always prefer asyncio.timeout() if you are using Python 3.11 or newer. It is the idiomatic, safe, and easiest way to handle timeouts in asyncio. For older projects, migrate to it when you can.