How to Get the Asyncio Task for a Coroutine

November 27, 2022 Python Asyncio

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.

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?

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.

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

Takeaways

You now know how to get an asyncio task for a coroutine.

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.