Sling Academy
Home/Python/Python asyncio.wait_for() function (with examples)

Python asyncio.wait_for() function (with examples)

Last updated: August 02, 2023

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!

Next Article: Python async/await and timeouts (with examples)

Previous Article: Python asyncio.run() function (with examples)

Series: Python Asynchronous Programming Tutorials

Python

You May Also Like

  • Python TypeError: write() argument must be str, not bytes
  • 4 ways to install Python modules on Windows without admin rights
  • Python TypeError: object of type ‘NoneType’ has no len()
  • Python: How to access command-line arguments (3 approaches)
  • Understanding ‘Never’ type in Python 3.11+ (5 examples)
  • Python: 3 Ways to Retrieve City/Country from IP Address
  • Using Type Aliases in Python: A Practical Guide (with Examples)
  • Python: Defining distinct types using NewType class
  • Using Optional Type in Python (explained with examples)
  • Python: How to Override Methods in Classes
  • Python: Define Generic Types for Lists of Nested Dictionaries
  • Python: Defining type for a list that can contain both numbers and strings
  • Using TypeGuard in Python (Python 3.10+)
  • Python: Using ‘NoReturn’ type with functions
  • Type Casting in Python: The Ultimate Guide (with Examples)
  • Python: Using type hints with class methods and properties
  • Python: Typing a function with default parameters
  • Python: Typing a function that can return multiple types
  • Python: Typing a function with *args and **kwargs