How to Get the Current Asyncio Task in Python

November 15, 2022 Python Asyncio

You can get the current task via asyncio.current_task() function.

In this tutorial, you will discover how to get and use the current 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:

...
# create and schedule a task
task = asyncio.create_task(coro)

You can learn more about asyncio tasks in the tutorial:

What is an Asyncio Task

Now that we know about asyncio tasks, let's look at how we might get the current task.

How to Get the Current Task

We can get the current task via the asyncio.current_task() function.

This function will return a Task object for the task that is currently running.

For example:

...
# get the current task
task = asyncio.current_task()

This will return a Task object for the currently running task.

This may be:

The main coroutine passed to asyncio.run().
A task created and scheduled within the asyncio program via asyncio.create_task().

A task may create and run another coroutine (e.g. not wrapped in a task). Getting the current task from within a coroutine will return a Task object for the running task, but not the coroutine that is currently running.

Getting the current task can be helpful if a coroutine or task requires details about itself, such as the task name for logging.

Now that we know how to get the current task, let's look at some worked examples.

Example of Getting the Current Task in the Main Coroutine

We can explore how to get a Task instance for the main coroutine used to start an asyncio program.

The example below defines a coroutine used as the entry point into the program. It reports a message, then gets the current task and reports its details.

This is an important first example, as it highlights that all coroutines can be accessed as tasks within the asyncio event loop.

The complete example is listed below.

# SuperFastPython.com
# example of getting the current task from the main coroutine
import asyncio

# define a main coroutine
async def main():
    # report a message
    print('main coroutine started')
    # get the current task
    task = asyncio.current_task()
    # report its details
    print(task)

# start the asyncio program
asyncio.run(main())

Running the example first creates the main coroutine and uses it to start the asyncio program.

The main() coroutine runs and first reports a message.

It then retrieves the current task, which is a Task object that represents itself, the currently running coroutine.

It then reports the details of the currently running task.

We can see that the task has the default name for the first task, 'Task-1' and is executing the main() coroutine, the currently running coroutine.

This highlights that we can use the asyncio.current_task() function to access a Task object for the currently running coroutine, that is automatically wrapped in a Task object.

main coroutine started
<Task pending name='Task-1' coro=<main() running at ...> cb=[_run_until_complete_cb() at ...]>

Next, let's take a look at how we might get a Task object from another coroutine.

Example of Getting the Current Task From Another Coroutine

The asyncio.current_task() function can also be used to retrieve a Task object for the current task from another coroutine that is currently running.

We can explore running a second coroutine in an asyncio program and getting the current Task object from within it.

In this example, the main coroutine is used as the entry point into the program. It is converted into a Task automatically.

The main coroutine creates and runs a second coroutine and this second coroutine retrieves the current task, which refers to the main coroutine, not the currently running coroutine.

The complete example is listed below.

# SuperFastPython.com
# example of getting the current task from another coroutine
import asyncio

# define another coroutine
async def another_coroutine():
    # report a message
    print('executing the coroutine')
    # get the current task
    task = asyncio.current_task()
    # report its details
    print(task)

# define a main coroutine
async def main():
    # report a message
    print('main coroutine started')
    # wait another coroutine
    await another_coroutine()
    # report a final message
    print('main coroutine done')

# start the asyncio program
asyncio.run(main())

Running the example creates the main() coroutine and uses it as the entry point into the asyncio program.

The main() coroutine reports a message, then creates and runs a second coroutine, then suspends waiting for it to be done.

The second coroutine runs, first reporting a message. It then gets the current Task object for itself and reports its details.

We can see that the asyncio.current_task() returns a Task instance for the main() coroutine that was wrapped in a task when it was used to start the program.

It does not refer to the coroutine that is currently running.

This highlights that the coroutine used as the entry point is turned into a task within the event loop, but arbitrary coroutines executing within a task are not themselves turned into a task and cannot be referenced directly as tasks.

main coroutine started
executing the coroutine
<Task pending name='Task-1' coro=<main() running at ...> cb=[_run_until_complete_cb() at ...]>
main coroutine done

Next, we will look at getting a task object for a new separate task.

Example of Getting the Current Task From Another Task

We can create and schedule a new task and the new task can get a Task object to reference itself.

In this example, we will define a coroutine that we will then wrap in a new Task. The task will get itself and report its details.

The complete example is listed below.

# SuperFastPython.com
# example of getting the current task from another task
import asyncio

# define another coroutine
async def another_coroutine():
    # report a message
    print('executing the task')
    # get the current task
    task = asyncio.current_task()
    # report its details
    print(task)

# define a main coroutine
async def main():
    # report a message
    print('main coroutine started')
    # create another task
    task = asyncio.create_task(another_coroutine())
    # await the task
    await task
    # report a final message
    print('main coroutine done')

# start the asyncio program
asyncio.run(main())

Running the example creates the main() coroutine and uses it as the entry point into the asyncio program.

The main() coroutine reports a message. It then creates and schedules a task and suspends it until it is done.

The new task runs. It reports a message, then gets a Task object for the currently running task and reports its details.

We can see that a new second task was created and used to execute the second coroutine. It has the name 'Task-2', compared to 'Task-1' for the entry point to the program that we saw in the above example. We can also see that the coroutine executed by the task matches our new second coroutine.

This highlights that we can get a Task object for new tasks that are created and scheduled within our program.

main coroutine started
executing the task
<Task pending name='Task-2' coro=<another_coroutine() running at ...> cb=[Task.task_wakeup()]>
main coroutine done

Next, let's look at what happens if a task attempts to await itself.

Example of a Task awaiting Itself

A task cannot await itself.

If we get the current task and then attempt to await it, a RuntimeError will be raised.

We can demonstrate this with a worked example.

# SuperFastPython.com
# example of getting the current task and waiting on itself
import asyncio

# define a main coroutine
async def main():
    # report a message
    print('main coroutine started')
    # get the current task
    task = asyncio.current_task()
    # attempt to wait on itself
    await task

# start the asyncio program
asyncio.run(main())

Running the example creates the main() coroutine and uses it as the entry point into the asyncio program.

The main() coroutine runs, first reporting a message.

It then gets the Task object for the currently running task, which refers to itself.

It then attempts to await this task, which is itself.

This fails with a RuntimeError and a message that a task cannot await itself.

This highlights that a task can interact with itself (introspection), but it cannot await itself.

main coroutine started
Traceback (most recent call last):
  ...
RuntimeError: Task cannot await on itself: <Task pending name='Task-1' coro=<main() running at ...> cb=[_run_until_complete_cb() at ...]>

Next, let's look at what happens if a task attempts to cancel itself.

Example of a Task Canceling Itself

Once a task gets a reference to itself, it can then query or perform operations on itself.

One operation a task may perform on itself is canceling the task via the cancel() method.

Recall that launching a task will raise a CancelledError exception in the running coroutine wrapped in the task.

You can learn more about canceling tasks in the tutorial:

How to Cancel an Asyncio Task

A task can cancel itself.

If a task calls the cancel() method on itself, it will in turn raise a CancelledError exception in the currently running coroutine.

The example below demonstrates this.

# SuperFastPython.com
# example of getting the current task and canceling itself
import asyncio

# define a main coroutine
async def main():
    # report a message
    print('main coroutine started')
    # get the current task
    task = asyncio.current_task()
    # attempt to cancel the task
    was_canceled = task.cancel()
    print(f'canceled: {was_canceled}')

Running the example creates the main() coroutine and uses it as the entry point into the asyncio program.

The main() coroutine runs, first reporting a message.

It then gets the Task object for the currently running task, which refers to itself.

The main() coroutine then calls the cancel() method on the Task object that represents itself.

This raises a CancelledError exception in the coroutine, causing the asyncio program to terminate.

This highlights that indeed a task can cancel itself.

main coroutine started
canceled: True
Traceback (most recent call last):
  ...
asyncio.exceptions.CancelledError

Takeaways

You now know how to get the current asyncio task in Python.

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.