close

How to schedule and cancel tasks with asyncio

Hello Guys, How are you all? Hope You all Are Fine. Today We Are Going To learn about How to schedule and cancel tasks with asyncio in Python. So Here I am Explain to you all the possible Methods here.

Without wasting your time, Let’s start This Article.

Table of Contents

How to schedule and cancel tasks with asyncio?

  1. How to schedule and cancel tasks with asyncio?

    ensure_future will automatically wrap your coroutine in a Task wrapper and attach it to your event loop.

  2. schedule and cancel tasks with asyncio

    ensure_future will automatically wrap your coroutine in a Task wrapper and attach it to your event loop.

Method 1

You can use asyncio Task wrappers to execute a task via the ensure_future() method.

ensure_future will automatically wrap your coroutine in a Task wrapper and attach it to your event loop. The Task wrapper will then also ensure that the coroutine ‘cranks-over’ from await to await statement (or until the coroutine finishes).

In other words, just pass a regular coroutine to ensure_future and assign the resultant Task object to a variable. You can then call Task.cancel() when you need to stop it.

import asyncio

async def task_func():
    print('in task_func')
    # if the task needs to run for a while you'll need an await statement
    # to provide a pause point so that other coroutines can run in the mean time
    await some_db_or_long_running_background_coroutine()
    # or if this is a once-off thing, then return the result,
    # but then you don't really need a Task wrapper...
    # return 'the result'

async def my_app():
    my_task = None
    while True:
        await asyncio.sleep(0)

        # listen for trigger / heartbeat
        if heartbeat and my_task is None:
            my_task = asyncio.ensure_future(task_func())

        # also listen for termination of hearbeat / connection
        elif not heartbeat and my_task:
            if not my_task.cancelled():
                my_task.cancel()
            else:
                my_task = None

run_app = asyncio.ensure_future(my_app())
event_loop = asyncio.get_event_loop()
event_loop.run_forever()

Note that tasks are meant for long-running tasks that need to keep working in the background without interrupting the main flow. If all you need is a quick once-off method, then just call the function directly instead.

Summery

It’s all About this issue. Hope all Methods helped you a lot. Comment below Your thoughts and your queries. Also, Comment below which Method worked for you? Thank You.

Also, Read