You can get the coroutine wrapped in a task by calling the get_coro() method on the Task object.
In this tutorial, you will discover how to get the coroutine from an asyncio task in Python.
Let’s get started.
What is an Asyncio Task
An asyncio Task is an object that schedules and independently runs an asyncio coroutine.
It provides a handle on a scheduled coroutine that an asyncio program can query and use to interact with the coroutine.
A Task is an object that manages an independently running coroutine.
— PEP 3156 – Asynchronous IO Support Rebooted: the “asyncio” Module
An asyncio task is represented via an instance of the asyncio.Task class.
A task is created from a coroutine. It requires a coroutine object, wraps the coroutine, schedules it for execution, and provides ways to interact with it.
A task is executed independently. This means it is scheduled in the asyncio event loop and will execute regardless of what else happens in the coroutine that created it. This is different from executing a coroutine directly, where the caller must wait for it to complete.
Tasks are used to schedule coroutines concurrently. When a coroutine is wrapped into a Task with functions like asyncio.create_task() the coroutine is automatically scheduled to run soon
— Coroutines and Tasks
We can create a task using the asyncio.create_task() function.
This function takes a coroutine instance and an optional name for the task and returns an asyncio.Task instance.
Wrap the coro coroutine into a Task and schedule its execution. Return the Task object.
— Coroutines and Tasks
For example:
|
1 2 3 |
... # create and schedule a task task = asyncio.create_task(coro) |
You can learn more about asyncio tasks in the tutorial:
Now that we know about asyncio tasks, let’s look at how we might get a coroutine from a task.
Run loops using all CPUs, download your FREE book to learn how.
How to get Coroutine from a Task
We can get a coroutine from a task via the get_coro() method on a Task object.
For example:
|
1 2 3 |
... # get the coroutine from a task coro = task.get_coro() |
Recall that a Task wraps a coroutine.
The coroutine that is wrapped can be retrieved again via the get_coro() method.
There is not a lot we can do with the coroutine once retrieved.
We can compare it with the coroutine passed to create_task() function.
If we try to reuse the coroutine or await the coroutine directly, an error will be raised.
Example of Getting a Coroutine from a Task
We can explore how to get the coroutine wrapped in a Task object.
In this example, we will create a coroutine and then wrap it in a task that will schedule it for execution.
Next, we will wait a moment, then retrieve the coroutine from the running task. We will report its details so we can compare it to the coroutine we wrapped in the task.
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 getting the coroutine wrapped in a task import asyncio # define a coroutine for a task async def task_coroutine(): # report a message print('executing the task') # block for a moment await asyncio.sleep(1) # custom coroutine async def main(): # report a message print('main coroutine started') # create a coroutine coro = task_coroutine() print(coro) # create and schedule the task from the coroutine task = asyncio.create_task(coro) # wait a moment await asyncio.sleep(0.1) # get the coroutine wrapped in the task retrieved_coro = task.get_coro() print(retrieved_coro) # check if it the same object as we wrapped is_same = coro is retrieved_coro print(f'same: {is_same}') # wait for the task to complete await task # report a final message print('main coroutine done') # start the asyncio program asyncio.run(main()) |
Running the example starts the asyncio event loop and executes the main() coroutine.
The main() coroutine reports a message, then creates a coroutine and reports its details, then wraps and schedules it as a task.
The main() coroutine then waits a moment to allow the task to begin executing.
The task runs, reports a message, then sleeps for a moment.
The main() coroutine resumes. It retrieves the coroutine from the task and reports its details.
We can see that the coroutine has the same memory address as the coroutine we created earlier. It is the same Python object.
We then compare the retrieved coroutine directly to the coroutine we wrapped in the task using the “is” keyword and report the result.
This confirms that indeed the retrieved coroutine is identical to the coroutine created earlier.
The main() coroutine then suspends and waits for the task to complete. Once finished, it reports a final message.
This highlights how we might retrieve a coroutine from a running task.
|
1 2 3 4 5 6 |
main coroutine started <coroutine object task_coroutine at 0x10812cdd0> executing the task <coroutine object task_coroutine at 0x10812cdd0> same: True main coroutine done |
Next, let’s look at what happens if we attempt to await a coroutine wrapped in 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 Awaiting a Coroutine from a Task
Once a coroutine is wrapped in a task, it cannot be awaited separately.
Doing so will result in a RuntimeError.
We can explore this case with an example.
In the example below, we create and schedule a coroutine in a task. The main coroutine waits a moment, then retrieves the wrapped coroutine from the running task and attempts to wait for it directly.
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 attempting to await a wrapped coroutine import asyncio # define a coroutine for a task async def task_coroutine(): # report a message print('executing the task') # block for a moment await asyncio.sleep(1) # custom coroutine async def main(): # report a message print('main coroutine started') # create and schedule the task from the coroutine task = asyncio.create_task(task_coroutine()) # wait a moment await asyncio.sleep(0.1) # get the coroutine wrapped in the task retrieved_coro = task.get_coro() # wait on the coroutine directly await retrieved_coro # report a final message print('main coroutine done') # start the asyncio program asyncio.run(main()) |
Running the example starts the asyncio event loop and executes the main() coroutine.
The main() coroutine reports a message, then creates and schedules the task coroutine.
It then waits a moment for the task to begin running.
The task runs, reports a message, and sleeps for a moment before terminating normally.
The main() coroutine resumes and retrieves the coroutine from the running task.
It then attempts to await it directly. Normally, this would execute the coroutine and wait for it to finish.
This fails with a RuntimeError indicating that the task is already being awaited.
This example highlights that once a task is used within a Task, it cannot be awaited again separately.
|
1 2 3 4 5 |
main coroutine started executing the task Traceback (most recent call last): ... RuntimeError: coroutine is being awaited already |
Next, let’s look at what happens if we try to wrap a coroutine from a task in another task.
Overwhelmed by the python concurrency APIs?
Find relief, download my FREE Python Concurrency Mind Maps
Example of Reusing a Coroutine from a Task
Once a coroutine is wrapped in a task, it cannot be reused.
This means that it cannot be awaited in a coroutine, as we saw in the previous example.
It also means that we cannot wrap it in another task object in order to re-schedule and execute it independently.
We can explore this case with a worked example.
In the example below, we wrap a coroutine in a task and let it begin executing. The coroutine is retrieved, then the task is awaited until it is complete. The retrieved coroutine is then used to create and schedule a second task, which is expected to fail with a RuntimeError.
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 attempting to reuse a coroutine import asyncio # define a coroutine for a task async def task_coroutine(): # report a message print('executing the task') # block for a moment await asyncio.sleep(1) # custom coroutine async def main(): # report a message print('main coroutine started') # create and schedule the task from the coroutine task = asyncio.create_task(task_coroutine()) # wait a moment await asyncio.sleep(0.1) # get the coroutine wrapped in the task retrieved_coro = task.get_coro() # wait for the task to finish await task # attempt to reuse the coroutine task2 = asyncio.create_task(retrieved_coro) # wait for the task to finish await task2 # report a final message print('main coroutine done') # start the asyncio program asyncio.run(main()) |
Running the example starts the asyncio event loop and executes the main() coroutine.
The main() coroutine reports a message, then creates and schedules the task coroutine.
It then waits a moment for the task to begin running.
The task runs, reports a message, and sleeps for a moment before terminating normally.
The main() coroutine resumes and retrieves the coroutine from the running task.
It then awaits the task and lets it finish.
The main() coroutine then attempts to create and schedule a second task using the coroutine retrieved from the first task.
This fails with a RuntimeError indicating that the task cannot be reused.
This example highlights that once a task is used within a Task, it cannot be reused within a second task.
|
1 2 3 4 5 6 |
main coroutine started executing the task Traceback (most recent call last): ... RuntimeError: cannot reuse already awaited coroutine ... |
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 get the coroutine from an asyncio task in Python.
Do you have any questions?
Ask your questions in the comments below and I will do my best to answer.
Photo by Meritt Thomas on Unsplash
Do you have any questions?