How to Get the Asyncio Task for a Coroutine

You can get an asyncio.Task for a coroutine by searching through all running tasks.

In this tutorial, you will discover how to get an asyncio task for a coroutine in Python.

Let’s get started.

Table of Contents

Need a Task For a Coroutine

A coroutine is a function or routine that can be suspended and resumed. As such it has more than one entry point.

The “async def” expression defines a coroutine.

Functions defined with async def syntax are always coroutine functions, even if they do not contain await or async keywords.
— Python Compound statements

For example:

# define a coroutine

async def custom_coro():

# ...

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.

This function takes a coroutine instance and an optional name for the task and returns an asyncio.Task instance.

You can learn more about asyncio tasks in the tutorial:

What is an Asyncio Task

Often in asyncio programming, we may need an asyncio.Task object for a given coroutine.

This may be for many reasons.

For example, we may execute the coroutine independently as a task and later require the result from the task in our program.

For example:

...

# create coroutine

coro = custom_coro()

# create and execute coroutine as a task

_ = asyncio.create_task(coro())

Another example is that we may have shielded the coroutine from cancellation via asyncio.shield() and an attempt was made to cancel it, meaning the coroutine continues executing and we require access to its status or result.

For example:

...

# create coroutine

coro = custom_coro()

try:

# execute coroutine, shielded from cancellation

await asyncio.shield(coro)

except asyncio.CancelledError:

# ...

You can learn more about shielding coroutines from cancellation in the tutorial:

How to Shield an Asyncio Task from Cancellation

If we have access to the coroutine, how can we get the associated asyncio.Task object?

Run loops using all CPUs, download your FREE book to learn how.

How to Get an asyncio.Task Object for a Coroutine

We can get an asyncio.Task object for a coroutine by searching through all running tasks.

This can be achieved by first getting a list of all currently running task objects via the asyncio.all_tasks() function.

For example:

...

# get all tasks

tasks = asyncio.all_tasks()

We can then call the get_coro() function on each to get the coroutine that the task is executing.

If the coroutine matches the coroutine object we are searching for, we have found the matching task object.

For example:

...

# search for task that is running a given coroutine

for task in tasks:

# check if task is running the coro

if task.get_coro() is coro:

# found it

If we do not locate the task object for a given coroutine object it may mean one of a few situations, such as:

The task is no longer running, e.g. we were too late.
A task was not created for the coroutine, e.g. we made a mistake.
A task was not created yet, e.g. it needs time to run.

This approach assumes that we keep a reference to the coroutine that is used as the basis for the task.

We can define a helper function to get an asyncio.Task or a given coroutine.

For example:

# returns the task for the given coroutine or none

def task_for_coro(coro):

# search for task that is running a given coroutine

for task in asyncio.all_tasks():

# check if task is running the coro

if task.get_coro() is coro:

return task

return None

Now that we know how to get a task for a given coroutine, let’s look at a worked example.

Download Now: Free Asyncio PDF Cheat Sheet

Example of Getting a Task for a Coroutine

We can explore how to get a task for a coroutine.

In this example we will define a task coroutine that reports a message, sleeps, then reports a final message.

The main coroutine will create the coroutine and then wraps the coroutine in a task and schedule it for execution. It then sleeps for a moment then gets the task object for the coroutine using the helper function developed in the previous section.

The complete example is listed below.

# SuperFastPython.com

# example of getting a task for a coroutine

import asyncio

# coroutine run in a task

async def task_coro():

# report a message

print('Task is running')

# block for a moment

await asyncio.sleep(1)

# report another message

print('Task is done')

# returns the task for the given coroutine or none

def task_for_coro(coro):

# search for task that is running a given coroutine

for task in asyncio.all_tasks():

# check if task is running the coro

if task.get_coro() is coro:

return task

return None

# main coroutine

async def main():

# create the coroutine

coro = task_coro()

# create and start the task

_ = asyncio.create_task(coro)

# wait a moment

await asyncio.sleep(0.2)

# get the task for the coroutine

task = task_for_coro(coro)

# report the task

print(f'Found: {task}')

# wait for the task to be done

await task

# start 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 then creates the task_coro() coroutine and uses it to create and schedule a task. The reference to the asyncio.Task object is not kept.

Next, the main() coroutine sleeps and is suspended, the task runs, reports a message, then sleeps.

The main() coroutine resumes and searches for the Task object for the coroutine. It is found as expected and the details are reported.

The main() coroutine then awaits the task directly, suspending until the task is done.

This example highlights how we can locate and use the asyncio.Task object for a coroutine, as long as the task is still running and that we have reference to the coroutine.

Task is running

Found: <Task pending name='Task-2' coro=<task_coro() running at ...> wait_for=<Future pending cb=[Task.task_wakeup()]>>

Task is done

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.

Learn more