Last Updated on December 13, 2023
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?
Run loops using all CPUs, download your FREE book to learn how.
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:
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:
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:
|
1 2 3 |
... # 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:
|
1 2 3 |
... # 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:
|
1 2 3 |
... # 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:
|
1 2 3 4 5 6 7 |
... # 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:
|
1 2 3 4 5 6 7 8 9 |
... # 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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
... # 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.
|
1 2 3 4 5 6 7 8 9 10 11 |
... # 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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
... # 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.
|
1 2 3 4 5 6 |
# 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.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
# 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.
|
1 2 3 |
... # start the event loop asyncio.run(main()) |
Tying this together, the complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 |
# 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.
|
1 |
Timeout waiting |
Next, let’s explore how we might perform the same operation using a task.
Free Python Asyncio Course
Download your FREE Asyncio PDF cheat sheet and get BONUS access to my free 7-day crash course on the Asyncio API.
Discover how to use the Python asyncio module including how to define, create, and run new coroutines and how to use non-blocking I/O.
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:
|
1 2 3 |
... # schedule the task running_task = asyncio.create_task(task(1)) |
We can also allow the task to start running.
|
1 2 3 |
... # 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.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
... # 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.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 |
# 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.
|
1 |
Timeout waiting |
Overwhelmed by the python concurrency APIs?
Find relief, download my FREE Python Concurrency Mind Maps
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.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
# 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.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 |
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.
|
1 2 3 4 |
... # 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.
|
1 2 3 4 5 6 7 8 9 |
... # 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.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 |
# 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.
|
1 |
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.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
... # 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.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 |
# 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.
|
1 |
100 |
Further Reading
This section provides additional resources that you may find helpful.
Python Asyncio Books
- Python Asyncio Mastery, Jason Brownlee (my book!)
- Python Asyncio Jump-Start, Jason Brownlee.
- Python Asyncio Interview Questions, Jason Brownlee.
- Asyncio Module API Cheat Sheet
I also recommend the following books:
- Python Concurrency with asyncio, Matthew Fowler, 2022.
- Using Asyncio in Python, Caleb Hattingh, 2020.
- asyncio Recipes, Mohamed Mustapha Tahrioui, 2019.
Guides
APIs
- asyncio — Asynchronous I/O
- Asyncio Coroutines and Tasks
- Asyncio Streams
- Asyncio Subprocesses
- Asyncio Queues
- Asyncio Synchronization Primitives
References
Takeaways
You now know how to use asyncio.timeout_at() for setting deadlines in the future.
Did I make a mistake? See a typo?
I’m a simple humble human. Correct me, please!
Do you have any additional tips?
I’d love to hear about them!
Do you have any questions?
Ask your questions in the comments below and I will do my best to answer.
Photo by Varla Scooter on Unsplash
Do you have any questions?