Python asyncio.wait_for() function (with examples)

Updated: August 2, 2023 By: Wolf Post a comment

Overview

Syntax & parameters

The asyncio.wait_for() function was added in Python 3.41, and it allows you to wait for an awaitable object (such as a coroutine or a task) to complete within a specified timeout. If the timeout expires before the awaitable object is done, the function will cancel it and raise an asyncio.TimeoutError exception.

Syntax:

asyncio.wait_for(awaitable, timeout, *, loop=None)

The parameters of the function are:

  • awaitable: The object that you want to wait for. It can be a coroutine or a task.
  • timeout: The maximum number of seconds to wait for the awaitable object to complete. It can be None for no timeout or a positive number.
  • loop: The event loop to use for the operation. If None, the default event loop will be used.

The asyncio.wait_for() function returns the result of the awaitable object that it waits for if it completes within the specified timeout.

When and when NOT to use asyncio.wait_for()?

The asyncio.wait_for() function is helpful when you want to limit the execution time of an asynchronous operation and handle the possible timeout gracefully. For example, you can use it to implement a network request with a timeout or to cancel a long-running task if the user interrupts it.

You shouldn’t use the asyncio.wait_for() function when you don’t care about the completion of the awaitable object, or when you want to wait for multiple awaitable objects at once. In those cases, you can use other functions from the asyncio module, such as asyncio.create_task(), asyncio.gather(), or asyncio.wait().

Examples

Wait for a coroutine with a timeout (basic)

This might be the simplest example of using the asyncio.wait_for() function you’ve ever seen:

# SlingAcademy.com
# This code uses Python 3.11.4

import asyncio


async def say_hello():
    # Simulate a slow operation
    await asyncio.sleep(5)
    print("Hello buddy. Welcome to Sling Academy!")


async def main():
    try:
        # Wait for the say_hello coroutine to finish within 3 seconds
        await asyncio.wait_for(say_hello(), 3)
    except asyncio.TimeoutError:
        # Handle the timeout
        print("Timeout. The operation took too long to complete.")


# Run the main coroutine
asyncio.run(main())

The following message will be printed after 3 seconds because the say_hello coroutine takes longer than that to finish (5 seconds):

Timeout. The operation took too long to complete.

Wait for a task with a timeout (basic)

This example will print numbers from 0 to 9, then “Cancelled” after 10 seconds:

# SlingAcademy.com
# This code uses Python 3.11.4

import asyncio


async def count():
    # Simulate a long-running task
    i = 0
    while True:
        print(i)
        i += 1
        await asyncio.sleep(1)


async def main():
    # Create a task from the count coroutine
    task = asyncio.create_task(count())
    try:
        # Wait for the task to finish within 10 seconds
        await asyncio.wait_for(task, timeout=10)
    except asyncio.TimeoutError:
        # Cancel the task
        task.cancel()
        print("Cancelled")


# Run the main coroutine
asyncio.run(main())

Output:

0
1
2
3
4
5
6
7
8
9
Cancelled

If we set timeout to None, the counting task will run forever, and you will never see “Cancelled”.

Wait for a coroutine with a timeout and handle the result (intermediate)

This example will print either “The result is X” where X is a random number, or “Timeout” after 5 seconds:

# SlingAcademy.com
# This code uses Python 3.11.4

import asyncio
import random


async def get_number():
    # Simulate a slow operation that returns a random number
    await asyncio.sleep(random.randint(1, 10))
    return random.randint(1, 100)


async def main():
    try:
        # Wait for the get_number coroutine to finish within 5 seconds
        result = await asyncio.wait_for(get_number(), 5)
        print(f"The result is {result}")
    except asyncio.TimeoutError:
        # Handle the timeout
        print("Timeout")


# Run the main coroutine
asyncio.run(main())

The result depends on how long the get_number coroutine takes to finish.

Afterword

We’ve walked through the fundamentals of the asyncio.wait_for() function and examined some examples of using the function in practice. If you find something outdated or incorrect, please let me know by leaving a comment. This tutorial ends here. Happy coding & go forth!