Use asyncio.timeout_at() to Run Tasks With Deadlines

December 12, 2023 Python Asyncio

You can wait for asyncio tasks with a deadline using the asyncio.timeout_at() context manager.

This asynchronous context manager will cancel the task if it takes too long and will raise an asyncio.Timeout exception, which can be handled to clean up after the task.

In this tutorial, you will discover how to use asyncio.timeout_at() for setting deadlines in the future.

Let's get started.

Need to Wait For Long-Running Task With a Timeout

We often need to execute long-running tasks in asyncio.

For example:

We may need to wait for a response from a remote server.
We may need to wait for something to change.
We may need to wait for input from a user or system.

It is a best practice to use a timeout when waiting for a long-running task.

If the task does not complete within a given time limit, can then be canceled and perhaps tried again, or an error raised.

The asyncio module provides the asyncio.timeout_at() asynchronous context manager to address exactly this problem.

How can we use asyncio.timeout_at() to add timeouts to long-running tasks?

How to Use asyncio.timeout_at()

The asyncio.timeout_at() is an asynchronous context manager.

As such, the entry and exit methods are coroutines that can be awaited and it must be used via the "async with" expression.

You can learn more about asynchronous context managers in the tutorial:

Asynchronous Context Managers in Python

The asyncio.timeout_at() context manager and associated asyncio.timeout() context manager and asyncio.Timeout class were added to Python in version 3.11.

You can learn more about the closely related asyncio.timeout() context manager in the tutorial:

Wait and Cancel Tasks With asyncio.timeout()

The timeout argument to asyncio.timeout_at() is "when", an absolute time in the future.

when is the absolute time to stop waiting, or None.
-- Coroutines and Tasks

A "when" time can be calculated by retrieving the current event loop time and adding a number of seconds to it, as either an integer or floating point value.

The current event loop can be retrieved via the asyncio.get_running_loop() function.

For example:

...
# get the event loop object
loop = asyncio.get_running_loop()

The current event loop time can be retrieved by calling the time() method on the event loop object.

For example:

...
# get the current event loop time
current_time = loop.time()

A future time for the "when" argument can be calculated by adding a fixed number of seconds to the current event loop time.

For example:

...
# calculate a deadline
deadline = current_time + 5

This can then be used in the asyncio.timeout_at() context manager to execute one or more asyncio tasks with the deadline.

For example:

...
# run tasks with a deadline
async with asyncio.timeout_at(deadline):
    # execute long running task
    result = await long_running_task()
    # process result
    # ...

The benefit of asyncio.timeout_at() being a context manager, means that we can await many coroutines, e.g. many sub-tasks, within the body and only cancel the one task that takes too long.

For example:

...
# run tasks with a deadline
async with asyncio.timeout_at(deadline):
    # execute many tasks
    await task1()
    await task2()
    await task3()
    await task4()
    await task5()

If the current time exceeds the provided deadline, the current task that is running is canceled, and an asyncio.TimeoutError exception is raised.

As such, this exception should be handled.

For example:

...
# get the event loop object
loop = asyncio.get_running_loop()
# get the current event loop time
current_time = loop.time()
# calculate a deadline
deadline = current_time + 5
# handle timeout
try:
	# run tasks with a deadline
	async with asyncio.timeout_at(deadline):
        # execute long running task
        result = await task(1)
        # report the result
        print(result)
except asyncio.TimeoutError:
    # handle timeout

The asyncio.timeout_at() function creates an instance of the asyncio.Timeout class which has three methods:

asyncio.Timeout.when(): Return the current deadline
asyncio.Timeout.reschedule(): Change the current deadline to a new time.
asyncio.Timeout.expired(): Return True if the timeout has expired, False otherwise.

It is possible to extend the timeout further into the future.

This can be used to either set no timeout initially, and then set a timeout later while running, or to extend a predefined timeout further into the future.

For example, we can set a timeout by passing None for the "when" argument, then calling the reschedule() method on the asynchronous context manager with a time in the future.

...
# set no timeout
async with asyncio.timeout_at(None) as timeout:
	# do something else
	# ...
	# calculate a deadline 5 seconds in the future
	deadline = asyncio.get_running_loop().time() + 5
	# set the new deadline
	timeout.reschedule(deadline)
    # execute long running task
    result = await task()

We can also set an initial timeout delay, then extend it into the future using the same approach.

For example:

...
# get the event loop object
loop = asyncio.get_running_loop()
# get the current event loop time
current_time = loop.time()
# calculate a deadline
deadline = current_time + 5
# set deadline
async with asyncio.timeout_at(deadline) as timeout:
	# do something else
	# ...
	# calculate a deadline 10 seconds in the future
	deadline = asyncio.get_running_loop().time() + 10
	# set the new deadline
	timeout.reschedule(deadline)
    # execute long running task
    result = await task()

Now that we know how to use asyncio.timeout_at(), let's look at some worked examples.

Example of How to Use asyncio.timeout_at()

We can explore an example of waiting for a long-running coroutine with a deadline.

In this example, we will define a long-running coroutine that takes an argument and returns a result. We will then execute this coroutine within an asyncio.timeout_at() context manager and handle any asyncio.TimeoutError raised.

Firstly, we can define the long-running coroutine. In this case, it sleeps for 10 seconds.

# long running task
async def task(value):
    # sleep to simulate waiting
    await asyncio.sleep(10)
    # return value
    return value * 100

Next, we can define the main coroutine.

It first handles the asyncio.TimeoutError, then calculates a deadline in the future relative to the event loop time. It then opens the asyncio.timeout_at() context manager and sets the absolute deadline in the future.

Within the context manager, the task() coroutine is called, passing an argument and retrieving the return value which is then printed.

# asyncio entry point
async def main():
    # handle timeout
    try:
        # calculate a deadline 5 seconds in the future
        deadline = asyncio.get_running_loop().time() + 5
        # wait with a deadline
        async with asyncio.timeout_at(deadline):
            # execute long running task
            result = await task(1)
            # report the result
            print(result)
    except asyncio.TimeoutError:
        print(f'Timeout waiting')

We can then start the asyncio event loop and execute the main() coroutine.

...
# start the event loop
asyncio.run(main())

Tying this together, the complete example is listed below.

# SuperFastPython.com
# example of waiting for a coroutine with asyncio.timeout_at()
import asyncio

# long running task
async def task(value):
    # sleep to simulate waiting
    await asyncio.sleep(10)
    # return value
    return value * 100

# asyncio entry point
async def main():
    # handle timeout
    try:
        # calculate a deadline 5 seconds in the future
        deadline = asyncio.get_running_loop().time() + 5
        # wait with a deadline
        async with asyncio.timeout_at(deadline):
            # execute long running task
            result = await task(1)
            # report the result
            print(result)
    except asyncio.TimeoutError:
        print(f'Timeout waiting')

# start the event loop
asyncio.run(main())

Running the example calculates a deadline. It then opens the asyncio.timeout_at() and sets the deadline.

The task() coroutine is started and runs, sleeping for 5 seconds.

After 5 seconds, the asyncio.timeout_at() context manager resumes and cancels the task() coroutine and raises an asyncio.TimeoutError.

This exception is handled and a failure message is reported.

This example highlights how we can execute a long-running coroutine with a deadline, then cancel the coroutine after the deadline has passed and handle the termination exception.

Timeout waiting

Next, let's explore how we might perform the same operation using a task.

Example of a Deadline with Long Running asyncio.Task

We can explore an example of waiting for a long-running asyncio.Task with a deadline.

In this example, we will update the above example to instead use an asyncio.Task.

We can create and schedule the task() coroutine as an asyncio.Task before the asyncio.timeout_at() block.

For example:

...
# schedule the task
running_task = asyncio.create_task(task(1))

We can also allow the task to start running.

...
# allow the task to run
await asyncio.sleep(0)

We can then open the context manager and await this already scheduled task, with a deadline.

...
# handle timeout
try:
    # calculate a deadline 5 seconds in the future
    deadline = asyncio.get_running_loop().time() + 5
    # set a deadline
    async with asyncio.timeout_at(deadline):
        # wait for the task to complete
        result = await running_task
        # report the result
        print(result)
except asyncio.TimeoutError:
    print(f'Timeout waiting')

Tying this together, the complete example is listed below.

# SuperFastPython.com
# example of waiting for a task with asyncio.timeout_at()
import asyncio

# long running task
async def task(value):
    # sleep to simulate waiting
    await asyncio.sleep(10)
    # return value
    return value * 100

# asyncio entry point
async def main():
    # schedule the task
    running_task = asyncio.create_task(task(1))
    # allow the task to run
    await asyncio.sleep(0)
    # handle timeout
    try:
        # calculate a deadline 5 seconds in the future
        deadline = asyncio.get_running_loop().time() + 5
        # set a deadline
        async with asyncio.timeout_at(deadline):
            # wait for the task to complete
            result = await running_task
            # report the result
            print(result)
    except asyncio.TimeoutError:
        print(f'Timeout waiting')

# start the event loop
asyncio.run(main())

Running the example first creates the task() coroutine and schedules it as a task.

It then suspends the main() coroutine and allows the task to start running, beginning its sleep.

The main() coroutine resumes and calculates a deadline, then opens the timeout context manager and awaits the task.

After 5 seconds, the asyncio.timeout_at() context manager resumes and cancels the asyncio.Task and raises an asyncio.TimeoutError.

This exception is handled and a failure message is reported.

This example highlights how we can wait for an already executing long-running asyncio.Task with a deadline, then cancel it after the timeout expires and handle the termination exception.

Timeout waiting

Example of Deadline without Handling asyncio.TimeoutError

We can explore an example of waiting for a long-running coroutine with a timeout without handling the asyncio.TimeoutError.

In this case, we will update the above example to wait for a coroutine with a timeout, except we will not handle the exception.

The coroutine will be canceled raising an asyncio.exceptions.CancelledError, which will be caught and re-raised as an asyncio.TimeoutError.

This can be achieved by simply removing the try-except.

The complete example with this change is listed below.

# SuperFastPython.com
# example of waiting for a coroutine with asyncio.timeout_at()
import asyncio

# long running task
async def task(value):
    # sleep to simulate waiting
    await asyncio.sleep(10)
    # return value
    return value * 100

# asyncio entry point
async def main():
    # calculate a deadline 5 seconds in the future
    deadline = asyncio.get_running_loop().time() + 5
    # set a deadline
    async with asyncio.timeout_at(deadline):
        # execute long running task
        result = await task(1)
        # report the result
        print(result)

# start the event loop
asyncio.run(main())

Running the example first calculates a deadline, then opens the asyncio.timeout_at() and sets the future deadline.

The task() coroutine is started and runs, sleeping for 5 seconds.

After 5 seconds, the asyncio.timeout_at() context manager resumes and cancels the task() coroutine and raises a TimeoutError exception.

In this case, the exception is not handled and bubbles up, causing the program to terminate.

We can see both the source CancelledError exception and the wrapping TimeoutError exception.

This example highlights how we can execute a long-running coroutine with a deadline, then cancel the coroutine after the deadline passes and not handle the termination exception.

Traceback (most recent call last):
  File ""...", line 19, in main
    result = await task(1)
             ^^^^^^^^^^^^^
  File ""...", line 8, in task
    await asyncio.sleep(10)
  File "".../asyncio/tasks.py", line 639, in sleep
    return await future
           ^^^^^^^^^^^^
asyncio.exceptions.CancelledError

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File ""...", line 24, in <module>
    asyncio.run(main())
  File ".../asyncio/runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "".../asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "".../asyncio/base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File ""...", line 17, in main
    async with asyncio.timeout_at(deadline):
  File "".../asyncio/timeouts.py", line 111, in __aexit__
    raise TimeoutError from exc_val
TimeoutError

Example of Timeout With A Deadline Time Set Later

We can explore an example of waiting for a long-running coroutine without an initial deadline, then adding a deadline later.

In this case, we can set no initial deadline, e.g. None.

...
# do not set a deadline
async with asyncio.timeout_at(None) as timeout:
	# ...

We can then perform some work and determine a deadline in the future and set this into the asyncio.timeout_at() context manager object, e.g. an instance of the asyncio.Timeout class.

...
# do not set a deadline
async with asyncio.timeout_at(None) as timeout:
    # wait a moment
    await asyncio.sleep(1)
    # calculate a deadline 5 seconds in the future
    deadline = asyncio.get_running_loop().time() + 5
    # set the new deadline
    timeout.reschedule(deadline)

This allows the timeout to be set later, on demand.

Tying this together, the complete example is listed below.

# SuperFastPython.com
# example of setting a deadline later with asyncio.timeout_at()
import asyncio

# long running task
async def task(value):
    # sleep to simulate waiting
    await asyncio.sleep(10)
    # return value
    return value * 100

# asyncio entry point
async def main():
    # handle timeout
    try:
        # do not set a deadline
        async with asyncio.timeout_at(None) as timeout:
            # wait a moment
            await asyncio.sleep(1)
            # calculate a deadline 5 seconds in the future
            deadline = asyncio.get_running_loop().time() + 5
            # set the new deadline
            timeout.reschedule(deadline)
            # execute long running task
            result = await task(1)
            # report the result
            print(result)
    except asyncio.TimeoutError:
        print(f'Timeout waiting')

# start the event loop
asyncio.run(main())

Running the example opens the asyncio.timeout_at() without a timeout.

The main() coroutine sleeps to simulate other work.

A deadline of 5 seconds in the future is then determined and the timeout is set into the asyncio.timeout_at() context manager object, e.g. an instance of the asyncio.Timeout class.

The task() coroutine is then issued and awaited.

After 5 seconds, the asyncio.timeout_at() context manager resumes and cancels the task() coroutine and raises an asyncio.TimeoutError.

The exception is handled and a message is reported.

This example highlights how we can set a deadline after we have entered the block and that this deadline has the normal effect of canceling the blocked task and raising an asyncio.TimeoutError exception.

Timeout waiting

Example of Extending a Deadline Further Into The Future

We can explore an example of waiting for a long-running coroutine with a deadline and then extending the deadline further into the future later.

In this case, we will update the example with the long-running coroutine. A first deadline will be set 5 seconds into the future on the context manager, not enough time for the task to complete.

The program will perform other tasks, and then increase the deadline by another 11 seconds into the future.

It will then execute the long-running task, which will have enough time to complete.

...
# calculate a deadline 5 seconds in the future
deadline = asyncio.get_running_loop().time() + 5
# set a deadline
async with asyncio.timeout_at(deadline) as timeout:
    # wait a moment
    await asyncio.sleep(4)
    # calculate a deadline further into the future
    deadline = asyncio.get_running_loop().time() + 11
    # set the new deadline
    timeout.reschedule(deadline)
    # execute long running task
    result = await task(1)
    # report the result
    print(result)

Tying this together, the complete example is listed below.

# SuperFastPython.com
# example of extending a deadline delay with asyncio.timeout_at()
import asyncio

# long running task
async def task(value):
    # sleep to simulate waiting
    await asyncio.sleep(10)
    # return value
    return value * 100

# asyncio entry point
async def main():
    # handle timeout
    try:
        # calculate a deadline 5 seconds in the future
        deadline = asyncio.get_running_loop().time() + 5
        # set a deadline
        async with asyncio.timeout_at(deadline) as timeout:
            # wait a moment
            await asyncio.sleep(4)
            # calculate a deadline further into the future
            deadline = asyncio.get_running_loop().time() + 11
            # set the new deadline
            timeout.reschedule(deadline)
            # execute long running task
            result = await task(1)
            # report the result
            print(result)
    except asyncio.TimeoutError:
        print(f'Timeout waiting')

# start the event loop
asyncio.run(main())

Running the example first calculates a deadline 5 seconds into the future, then opens the asyncio.timeout_at() and sets the deadline, which will not be long enough for our task.

The main() coroutine sleeps for 4 seconds to simulate other work.

The main() coroutine resumes and updates the deadline to be 11 seconds into the future. This is enough time to complete the task.

The task() coroutine is then issued and awaited.

The task completes normally and the return value is reported.

This example highlights how we can set a deadline on the context manager, then update that deadline later as more information becomes available.

Takeaways

You now know how to use asyncio.timeout_at() for setting deadlines in the future.

If you enjoyed this tutorial, you will love my book: Python Asyncio Jump-Start. It covers everything you need to master the topic with hands-on examples and clear explanations.