You can run an asyncio coroutine after a delay by wrapping it in a coroutine that waits before executing the target coroutine.
In this tutorial, you will discover how to run a coroutine after a delay with asyncio in Python.
Let’s get started.
Need to Run a Delayed Task
A coroutine can schedule another coroutine or task for execution.
The scheduled coroutine or task will execute as soon as it is able within the asyncio event loop.
This can be achieved by suspending and awaiting the coroutine directly, or wrapping the coroutine in a task and executing it independently.
When developing asyncio programs, we may need to schedule a coroutine and have it begin executing after a delay.
The high-level asyncio does not provide the capability to schedule a coroutine to execute later, after a delay.
How can we run a coroutine after a delay in asyncio?
Run loops using all CPUs, download your FREE book to learn how.
How to Run a Delayed Task
We can develop a custom wrapper coroutine to execute a target coroutine after a delay.
The wrapper coroutine may take two arguments, a coroutine and a time in seconds.
It will sleep for the given delay interval in seconds, then await the provided coroutine.
The delay() coroutine below implements this.
|
1 2 3 4 5 6 |
# coroutine that will start another coroutine after a delay in seconds async def delay(coro, seconds): # suspend for a time limit in seconds await asyncio.sleep(seconds) # execute the other coroutine await coro |
To use the wrapper coroutine, a coroutine object can be created and either awaited directly or executed independently as a task.
For example, the caller may suspend and schedule the delayed coroutine and wait for it to be done:
|
1 2 3 |
... # execute a coroutine after a delay await delay(coro, 10) |
Alternatively, the caller may schedule the delayed coroutine to run independently:
|
1 2 3 |
... # execute a coroutine after a delay independently _ = asyncio.create_task(delay(coro, 10)) |
Now that we know how to execute a coroutine after a delay, let’s look at some worked examples.
Example of Running a Delayed Coroutine
We can explore how to execute a coroutine after a delay.
In this example, we will define a coroutine task that reports a message, simulates work with a sleep, then reports a final message.
The main coroutine will execute the task coroutine after a delay and directly wait for it to complete.
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 |
# SuperFastPython.com # example of starting a coroutine after a delay import asyncio # coroutine that will start another coroutine after a delay in seconds async def delay(coro, seconds): # suspend for a time limit in seconds await asyncio.sleep(seconds) # execute the other coroutine await coro # task coroutine to do something async def task_coro(): # report a message print('Coroutine is running') # simulate some long running task await asyncio.sleep(3) # report another message print('Coroutine is done') # main coroutine async def main(): # report a message print('Main starting the delay') # run the delayed coroutine await delay(task_coro(), 3) # report a final message print('Main all done') # run the asyncio program asyncio.run(main()) |
Running the example first creates the main() coroutine and uses it as the entry point into the asyncio program.
The main() coroutine runs and reports a message.
It then creates the task_coro() coroutine and wraps it in the delay() coroutine. The main() coroutine then suspends and awaits the delayed coroutine to complete.
The delay() coroutine runs and suspends for the given time limit, in this case, three seconds.
It then resumes and executes the provided coroutine, which in this case is the task_coro() coroutine.
The task_coro() coroutine runs, reporting a message, sleeping for 3 seconds, and reporting a final message before terminating.
The delay() coroutine then terminates and the main() coroutine resumes and reports a final message.
This highlights how we can execute a coroutine after a delay.
|
1 2 3 4 |
Main starting the delay Coroutine is running Coroutine is done Main all done |
Next, let’s look at how we might execute a coroutine after a delay by scheduling it as an independent 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 Running a Delayed Task
We can explore how to run a coroutine after a delay by scheduling it as an independent task.
In this example, we will wrap our target task coroutine in the delay coroutine to ensure it is not started until after a delay. We will then schedule the delay wrapper coroutine to execute independently as a task.
The main coroutine will then sleep a moment to allow the scheduled task to execute. It will then simulate proceeding with other tasks, in this case sleeping long enough to allow the delayed coroutine to complete.
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 35 |
# SuperFastPython.com # example of starting a coroutine after a delay import asyncio # coroutine that will start another coroutine after a delay in seconds async def delay(coro, seconds): # suspend for a time limit in seconds await asyncio.sleep(seconds) # execute the other coroutine await coro # task coroutine to do something async def task_coro(): # report a message print('Coroutine is running') # simulate some long running task await asyncio.sleep(3) # report another message print('Coroutine is done') # main coroutine async def main(): # report a message print('Main starting the delay') # schedule the delayed task task = asyncio.create_task(delay(task_coro(), 3)) # start the task executing await asyncio.sleep(0) # simulate doing other things... await asyncio.sleep(7) # report a final message print('Main all done') # run the asyncio program asyncio.run(main()) |
Running the example first creates the main() coroutine and uses it as the entry point into the asyncio program.
The main() coroutine runs and reports a message.
It then creates the task_coro() coroutine and wraps it in the delay() coroutine which in turn is wrapped in an asyncio.Task. This schedules the delayed coroutine for independent execution.
The main() coroutine then suspends for a moment and allows the scheduled coroutine to run.
The delay() coroutine runs and suspends for the given time limit, in this case, three seconds.
The main() coroutine resumes and continues on with other activities. In this case, it simply sleeps for seven seconds.
The task_coro() coroutine runs after the delay, reporting a message, sleeping for 3 seconds, and reporting a final message before terminating. The delay() coroutine then terminates.
The main() coroutine resumes after its sleep, reports a message, and then terminates.
This highlights how we can execute a coroutine after a delay independently of the calling coroutine.
|
1 2 3 4 |
Main starting the delay Coroutine is running Coroutine is done Main all done |
Overwhelmed by the python concurrency APIs?
Find relief, download my FREE Python Concurrency Mind Maps
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 run a coroutine after a delay.
Do you have any questions?
Ask your questions in the comments below and I will do my best to answer.
Photo by Ashley Byrd on Unsplash
it doesn`t work if period more than 3 seconds
task = asyncio.create_task(delay(task_coro(), 10))The first example will work with any amount of delay.
For example:
Does that help?
Running a delayed task as shown right now is tricky, because in case of cancellation during sleep, we are left with a never awaited
coro… which should count as a bug.Let’s follow a slightly modified example:
python# SuperFastPython.com
# example of starting a coroutine after a delay
import asyncio
# coroutine that will start another coroutine after a delay in seconds
async def delay(coro, seconds):
# suspend for a time limit in seconds
await asyncio.sleep(seconds)
# execute the other coroutine
await coro
# task coroutine to do something
async def task_coro():
# report a message
print('Coroutine is running')
# simulate some long running task
await asyncio.sleep(3)
# report another message
print('Coroutine is done')
# main coroutine
async def main():
# report a message
print('Main starting the delay')
# schedule the delayed task
task = asyncio.create_task(delay(task_coro(), 3))
# simulating some work
await asyncio.sleep(1)
# report the cancel
print("Main canceling the task")
# cancel the task gracefully
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
# report a final message
print('Main all done')
# run the asyncio program
asyncio.run(main())
Execution will result in the following output:
Main starting the delay
Main canceling the task
Main all done
RuntimeWarning: coroutine 'task_coro' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
Using [coroutine.close()](https://docs.python.org/3/reference/datamodel.html#coroutine.close) fixes the problem, but I’m not sure if it is robust enough to cover all
corostates:python# SuperFastPython.com
# example of starting a coroutine after a delay
import asyncio
# coroutine that will start another coroutine after a delay in seconds
async def delay(coro, seconds):
# suspend for a time limit in seconds
try:
await asyncio.sleep(seconds)
except asyncio.CancelledError:
# close not awaited coro in case of early cancellation
coro.close()
# re-rise the exception to comply with default behavior
raise
# execute the other coroutine
await coro
# task coroutine to do something
async def task_coro():
# report a message
print('Coroutine is running')
# simulate some long running task
await asyncio.sleep(3)
# report another message
print('Coroutine is done')
# main coroutine
async def main():
# report a message
print('Main starting the delay')
# schedule the delayed task
task = asyncio.create_task(delay(task_coro(), 3))
# simulating some work
await asyncio.sleep(1)
# report the cancel
print("Main canceling the task")
# cancel the task gracefully
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
# report a final message
print('Main all done')
# run the asyncio program
asyncio.run(main())
A task being canceled during a sleep is a feature.
You could perhaps overcome this using a shield for the deelayed task, e.g. asyncio.shield(delayed())