Метод cancel() модуля asyncio в Python — отмена задачи

D этом руководстве вы узнаете, как отменить длительную асинхронную операцию в Python, выполнение которой может занять целую вечность, с помощью метода cancel() объекта asyncio.Task.

В следующем примере используется оператор await для ожидания завершения задачи:

task = asyncio.create_task(coroutine())

result = await task

Однако, если coroutine() будет длиться вечно, вы застрянете в ожидании завершения оператора await, не получив никакого результата. Кроме того, у вас нет возможности остановить это, когда вы пожелаете.

Чтобы решить эту проблему, вы можете отменить задачу, используя метод cancel() объекта Task. Если вы отмените задачу, она вызовет исключение CancelledError, когда вы ее ожидаете. Например:

import asyncio
from asyncio import CancelledError


async def call_api(message, result=1000, delay=3):
    print(message)
    await asyncio.sleep(delay)
    return result


async def main():
    task = asyncio.create_task(
        call_api('Calling API...', result=2000, delay=5)
    )

    if not task.done():
        print('Cancelling the task...')
        task.cancel()

    try:
        await task
    except CancelledError:
        print('Task has been cancelled.')


asyncio.run(main())

Выход:

Cancelling the task...
Task has been cancelled

Как это работает.

• Сначала сопрограмма call_api() печатает сообщение, задерживается на 3 секунды и возвращает результат.
• Во-вторых, создайте новую задачу с помощью функции create_task() и передайте сопрограмму call_api(). Выполнение задачи займет 5 секунд:

task = asyncio.create_task(
    call_api('Calling API...', result=2000, delay=5)
)

• В-третьих, проверьте, не выполнена ли задача, вызвав метод Done(), и отмените задачу, используя метод cancel():

if not task.done():
    print('Cancelling the task...')
    task.cancel()

• Наконец, дождитесь завершения задачи, используя ключевое слово await. Поскольку задача была отменена, возникает исключение CancelledError:

try:
    await task
except CancelledError:
    print('Task has been cancelled.')

Если вы хотите проверять, выполнена ли задача, и отменить ее, если прошло определенное время, вы можете использовать цикл while:

import asyncio
from asyncio import CancelledError


async def call_api(message, result=1000, delay=3):
    print(message)
    await asyncio.sleep(delay)
    return result


async def main():
    task = asyncio.create_task(
        call_api('Calling API...', result=2000, delay=5)
    )

    time_elapsed = 0
    while not task.done():
        time_elapsed += 1
        await asyncio.sleep(1)
        print('Task has not completed, checking again in a second')
        if time_elapsed == 3:
            print('Cancelling the task...')
            task.cancel()
            break

    try:
        await task
    except CancelledError:
        print('Task has been cancelled.')


asyncio.run(main())

В этом примере цикл while проверяет, выполнена ли задача каждую секунду, и отменяет задачу, как только прошедшее время достигнет 3 секунд.