Сопрограммы и задачи

Сопрограммы

Исходный код: Lib/asyncio/coroutines.py

Сопрограммы, объявленные с помощью синтаксиса async/await, — предпочтительный способ написания приложений asyncio. Например, следующий фрагмент кода выводит «привет», ждёт 1 секунду, а затем выводит «мир»:

>>> import asyncio

>>> async def main():
...     print('hello')
...     await asyncio.sleep(1)
...     print('world')

>>> asyncio.run(main())
hello
world

Обратите внимание: простой вызов сопрограммы не планирует её выполнение:

>>> main()
<coroutine object main at 0x1053bb7c8>

Для фактического запуска сопрограммы asyncio предоставляет следующие механизмы:

• Функция asyncio.run() для запуска функции «main()», являющейся точкой входа верхнего уровня (см. пример выше).

• Ожидание сопрограммы. Следующий фрагмент кода выведет «привет» после ожидания в течение 1 секунды, а затем выведет «мир» после ожидания в течение ещё 2 секунд:

  import asyncio
  import time
  
  async def say_after(delay, what):
      await asyncio.sleep(delay)
      print(what)
  
  async def main():
      print(f"started at {time.strftime('%X')}")
  
      await say_after(1, 'hello')
      await say_after(2, 'world')
  
      print(f"finished at {time.strftime('%X')}")
  
  asyncio.run(main())
  

  Ожидаемый результат:

  started at 17:13:52
  hello
  world
  finished at 17:13:55
  

• Функция asyncio.create_task() для одновременного запуска сопрограмм в виде объектов asyncio Tasks.

  Изменим предыдущий пример и запустим две сопрограммы say_after одновременно:

  async def main():
      task1 = asyncio.create_task(
          say_after(1, 'hello'))
  
      task2 = asyncio.create_task(
          say_after(2, 'world'))
  
      print(f"started at {time.strftime('%X')}")
  
      # Wait until both tasks are completed (should take
      # around 2 seconds.)
      await task1
      await task2
  
      print(f"finished at {time.strftime('%X')}")
  

  Обратите внимание: ожидаемый результат показывает, что теперь фрагмент выполняется на 1 секунду быстрее, чем раньше:

  started at 17:14:32
  hello
  world
  finished at 17:14:34
  

• Класс asyncio.TaskGroup предоставляет более современную альтернативу create_task(). С использованием этого API предыдущий пример будет выглядеть так:

  async def main():
      async with asyncio.TaskGroup() as tg:
          task1 = tg.create_task(
              say_after(1, 'hello'))
  
          task2 = tg.create_task(
              say_after(2, 'world'))
  
          print(f"started at {time.strftime('%X')}")
  
      # The await is implicit when the context manager exits.
  
      print(f"finished at {time.strftime('%X')}")
  

  Время выполнения и результат должны быть такими же, как в предыдущем варианте.

  Добавлено в версии 3.11: asyncio.TaskGroup.

Объекты, допускающие ожидание

Объект называется объектом, допускающим ожидание, если его можно использовать в выражении await. Многие API asyncio предназначены для работы с такими объектами.

Существует три основных типа объектов, допускающих ожидание: сопрограммы, задачи и объекты Future.

Сопрограммы

Сопрограммы Python допускают ожидание, поэтому их можно ожидать из других сопрограмм:

import asyncio

async def nested():
    return 42

async def main():
    # Nothing happens if we just call "nested()".
    # A coroutine object is created but not awaited,
    # so it *won't run at all*.
    nested()  # will raise a "RuntimeWarning".

    # Let's do it differently now and await it:
    print(await nested())  # will print "42".

asyncio.run(main())

Задачи

Задачи используются для одновременного планирования сопрограмм.

Когда сопрограмма оборачивается в задачу с помощью таких функций, как asyncio.create_task(), её выполнение автоматически планируется на ближайшее время:

import asyncio

async def nested():
    return 42

async def main():
    # Schedule nested() to run soon concurrently
    # with "main()".
    task = asyncio.create_task(nested())

    # "task" can now be used to cancel "nested()", or
    # can simply be awaited to wait until it is complete:
    await task

asyncio.run(main())

Объекты Future

Future — это специальный низкоуровневый объект, допускающий ожидание, который представляет будущий результат асинхронной операции.

Ожидание объекта Future означает, что сопрограмма будет ждать, пока объект Future не будет завершён в другом месте.

Объекты Future в asyncio нужны, чтобы код, основанный на обратных вызовах, можно было использовать с async/await.

Как правило, нет необходимости создавать объекты Future на уровне кода приложения.

Объекты Future, иногда предоставляемые библиотеками и некоторыми API asyncio, можно ожидать:

async def main():
    await function_that_returns_a_future_object()

    # this is also valid:
    await asyncio.gather(
        function_that_returns_a_future_object(),
        some_python_coroutine()
    )

Хороший пример низкоуровневой функции, возвращающей объект Future, — loop.run_in_executor().

Создание задач

Исходный код: Lib/asyncio/tasks.py

asyncio.create_task(coro, *, name=None, context=None)

Оборачивает сопрограмму coro coroutine в Task и планирует её выполнение. Возвращает объект Task.

Если name не равен None, это значение задаётся в качестве имени задачи с помощью Task.set_name().

Необязательный аргумент context, который можно передавать только по имени, позволяет указать пользовательский contextvars.Context, в котором будет выполняться coro. Если context не указан, создаётся копия текущего контекста.

Задача выполняется в цикле, возвращаемом функцией get_running_loop(). Если в текущем потоке нет работающего цикла, возникает исключение RuntimeError.

Примечание

asyncio.TaskGroup.create_task() — новая альтернатива, основанная на структурированной конкурентности; она позволяет дожидаться выполнения группы связанных задач с надёжными гарантиями безопасности.

Важно

Сохраните ссылку на результат этой функции, чтобы задача не исчезла во время выполнения. Цикл событий хранит только слабые ссылки на задачи. Задача, на которую больше нигде нет ссылок, может быть удалена сборщиком мусора в любой момент, даже до завершения. Чтобы надёжно запускать фоновые задачи без ожидания их результата, соберите их в коллекцию:

background_tasks = set()

for i in range(10):
    task = asyncio.create_task(some_coro(param=i))

    # Add task to the set. This creates a strong reference.
    background_tasks.add(task)

    # To prevent keeping references to finished tasks forever,
    # make each task remove its own reference from the set after
    # completion:
    task.add_done_callback(background_tasks.discard)

Добавлено в версии 3.7.

Изменено в версии 3.8: Добавлен параметр name.

Изменено в версии 3.11: Добавлен параметр context.

Отмена задач

Задачи можно легко и безопасно отменять. При отмене задачи при первой же возможности в ней возникает исключение asyncio.CancelledError.

Рекомендуется использовать в сопрограммах блоки try/finally для надёжного выполнения действий по очистке. Если исключение asyncio.CancelledError перехватывается явно, обычно его следует повторно возбудить после завершения очистки. Исключение asyncio.CancelledError напрямую наследуется от BaseException, поэтому большинству кода не нужно учитывать его отдельно.

Компоненты asyncio, обеспечивающие структурированную конкурентность, например asyncio.TaskGroup и asyncio.timeout(), используют отмену во внутренних механизмах и могут работать неправильно, если сопрограмма подавляет исключение asyncio.CancelledError. Аналогично, пользовательскому коду, как правило, не следует вызывать uncancel. Однако если подавление исключения asyncio.CancelledError действительно необходимо, нужно также вызвать uncancel(), чтобы полностью сбросить состояние отмены.

Группы задач

Группы задач объединяют API создания задач с удобным и надёжным способом дождаться завершения всех задач в группе.

class asyncio.TaskGroup

Асинхронный менеджер контекста, содержащий группу задач. Задачи можно добавлять в группу с помощью create_task(). При выходе из менеджера контекста выполняется ожидание всех задач.

Добавлено в версии 3.11.

create_task(coro, *, name=None, context=None)

Создаёт задачу в этой группе задач. Сигнатура совпадает с сигнатурой asyncio.create_task().

Пример:

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(some_coro(...))
        task2 = tg.create_task(another_coro(...))
    print(f"Both tasks have completed now: {task1.result()}, {task2.result()}")

Инструкция async with будет ждать завершения всех задач в группе. Пока выполняется ожидание, в группу всё ещё можно добавлять новые задачи (например, передав tg одной из сопрограмм и вызывая tg.create_task() в этой сопрограмме). После завершения последней задачи и выхода из блока async with добавлять новые задачи в группу нельзя.

Когда одна из задач группы впервые завершается с исключением, отличным от asyncio.CancelledError, остальные задачи группы отменяются. После этого добавлять задачи в группу нельзя. Если в этот момент тело инструкции async with ещё выполняется (то есть __aexit__() ещё не был вызван), отменяется также задача, непосредственно содержащая инструкцию async with. Возникшее исключение asyncio.CancelledError прервёт выполнение await, но не выйдет за пределы содержащей его инструкции async with.

После завершения всех задач, если какие-либо из них завершились с исключением, отличным от asyncio.CancelledError, эти исключения объединяются в ExceptionGroup или BaseExceptionGroup (в зависимости от ситуации; см. соответствующую документацию), после чего возникает это исключение.

Два базовых исключения обрабатываются особым образом: если какая-либо задача завершается с исключением KeyboardInterrupt или SystemExit, группа задач всё равно отменяет остальные задачи и дожидается их завершения, но затем повторно возбуждается исходное исключение KeyboardInterrupt или SystemExit, а не ExceptionGroup или BaseExceptionGroup.

Если тело инструкции async with завершается исключением (то есть __aexit__() вызывается при установленном исключении), это обрабатывается так же, как сбой одной из задач: остальные задачи отменяются, после чего ожидается их завершение, а исключения, не связанные с отменой, объединяются в группу исключений и возбуждаются. Исключение, переданное в __aexit__(), также включается в группу исключений, если только это не asyncio.CancelledError. Для KeyboardInterrupt и SystemExit действует такое же исключение из правил, как и в предыдущем абзаце.

import asyncio
from asyncio import TaskGroup

class TerminateTaskGroup(Exception):
    """Exception raised to terminate a task group."""

async def force_terminate_task_group():
    """Used to force termination of a task group."""
    raise TerminateTaskGroup()

async def job(task_id, sleep_time):
    print(f'Task {task_id}: start')
    await asyncio.sleep(sleep_time)
    print(f'Task {task_id}: done')

async def main():
    try:
        async with TaskGroup() as group:
            # spawn some tasks
            group.create_task(job(1, 0.5))
            group.create_task(job(2, 1.5))
            # sleep for 1 second
            await asyncio.sleep(1)
            # add an exception-raising task to force the group to terminate
            group.create_task(force_terminate_task_group())
    except* TerminateTaskGroup:
        pass

asyncio.run(main())

Task 1: start
Task 2: start
Task 1: done

Ожидание

async asyncio.sleep(delay, result=None)

Приостанавливает выполнение на delay секунд.

Если задан result, он возвращается вызывающему коду после завершения сопрограммы.

Функция sleep() всегда приостанавливает текущую задачу, позволяя выполняться другим задачам.

Значение задержки 0 обеспечивает оптимизированный способ дать возможность выполняться другим задачам. Его можно использовать в длительно выполняющихся функциях, чтобы не блокировать цикл событий на всё время вызова функции.

Пример сопрограммы, которая выводит текущую дату каждую секунду в течение 5 секунд:

import asyncio
import datetime

async def display_date():
    loop = asyncio.get_running_loop()
    end_time = loop.time() + 5.0
    while True:
        print(datetime.datetime.now())
        if (loop.time() + 1.0) >= end_time:
            break
        await asyncio.sleep(1)

asyncio.run(display_date())

Изменено в версии 3.10: Удалён параметр loop.

Одновременное выполнение задач

awaitable asyncio.gather(*aws, return_exceptions=False)

Одновременно запускает объекты, допускающие ожидание, из последовательности aws.

Если какой-либо объект из aws является сопрограммой, она автоматически планируется как задача.

Если выполнение всех объектов, допускающих ожидание, завершается успешно, результатом будет общий список возвращённых значений. Порядок значений в результате соответствует порядку объектов в aws.

Если return_exceptions равно False (значение по умолчанию), первое возникшее исключение немедленно передаётся задаче, ожидающей gather(). Остальные объекты в последовательности aws не отменяются и продолжают выполняться.

Если return_exceptions равно True, исключения обрабатываются так же, как успешные результаты, и включаются в результирующий список.

Если gather() отменяется, все переданные объекты, допускающие ожидание, выполнение которых ещё не завершено, также отменяются.

Если какой-либо Task или Future из последовательности aws отменяется, это обрабатывается так, как если бы он вызвал исключение CancelledError; вызов gather() в этом случае не отменяется. Это предотвращает отмену других задач и объектов Future при отмене одной из переданных задач или объектов Future.

Примечание

Новая альтернатива для одновременного создания и запуска задач с ожиданием их завершения — asyncio.TaskGroup. TaskGroup обеспечивает более надёжные гарантии безопасности, чем gather, при планировании вложенных подзадач: если задача (или подзадача, то есть задача, запланированная другой задачей) вызывает исключение, TaskGroup отменяет остальные запланированные задачи, а gather — нет).

Пример:

import asyncio

async def factorial(name, number):
    f = 1
    for i in range(2, number + 1):
        print(f"Task {name}: Compute factorial({number}), currently i={i}...")
        await asyncio.sleep(1)
        f *= i
    print(f"Task {name}: factorial({number}) = {f}")
    return f

async def main():
    # Schedule three calls *concurrently*:
    L = await asyncio.gather(
        factorial("A", 2),
        factorial("B", 3),
        factorial("C", 4),
    )
    print(L)

asyncio.run(main())

# Expected output:
#
#     Task A: Compute factorial(2), currently i=2...
#     Task B: Compute factorial(3), currently i=2...
#     Task C: Compute factorial(4), currently i=2...
#     Task A: factorial(2) = 2
#     Task B: Compute factorial(3), currently i=3...
#     Task C: Compute factorial(4), currently i=3...
#     Task B: factorial(3) = 6
#     Task C: Compute factorial(4), currently i=4...
#     Task C: factorial(4) = 24
#     [2, 6, 24]

Примечание

Если return_exceptions равно false, отмена gather() после того, как вызов уже помечен как завершённый, не отменяет переданные объекты, допускающие ожидание. Например, gather может быть помечен как завершённый после передачи исключения вызывающему коду. Поэтому вызов gather.cancel() после перехвата исключения (возникшего в одном из объектов, допускающих ожидание) из gather не отменит остальные такие объекты.

Изменено в версии 3.7: Если отменяется сам gather, отмена передаётся дальше независимо от значения return_exceptions.

Изменено в версии 3.10: Удалён параметр loop.

Устарело с версии 3.10: Выдаётся предупреждение об устаревании, если не указаны позиционные аргументы либо не все позиционные аргументы являются объектами, подобными Future, и при этом нет работающего цикла событий.

Фабрика для немедленного запуска задач

asyncio.eager_task_factory(loop, coro, *, name=None, context=None)

Фабрика задач для их немедленного выполнения.

При использовании этой фабрики (через loop.set_task_factory(asyncio.eager_task_factory)) сопрограммы начинают выполняться синхронно во время создания Task. Задачи планируются в цикле событий только в том случае, если они блокируются. Это может повысить производительность, поскольку позволяет избежать накладных расходов на планирование циклом для сопрограмм, которые завершаются синхронно.

Типичный пример, когда это полезно, — сопрограммы, использующие кэширование или мемоизацию, чтобы по возможности избежать фактического ввода-вывода.

Примечание

Немедленное выполнение сопрограммы меняет семантику. Если сопрограмма возвращает результат или вызывает исключение, задача никогда не планируется в цикле событий. Если выполнение сопрограммы блокируется, задача планируется в цикле событий. Это изменение может повлиять на поведение существующих приложений. Например, вероятно, изменится порядок выполнения задач приложения.

Добавлено в версии 3.12.

asyncio.create_eager_task_factory(custom_task_constructor)

Создаёт фабрику задач для немедленного запуска, подобную eager_task_factory(), которая при создании новой задачи использует предоставленный custom_task_constructor вместо стандартного Task.

custom_task_constructor должен быть вызываемым объектом с сигнатурой, совпадающей с сигнатурой Task.__init__. Вызываемый объект должен возвращать совместимый с asyncio.Task объект.

Эта функция возвращает вызываемый объект, предназначенный для использования в качестве фабрики задач цикла событий через loop.set_task_factory(factory)).

Добавлено в версии 3.12.

Защита от отмены

awaitable asyncio.shield(aw)

Защищает объект, допускающий ожидание, от cancelled.

Если aw является сопрограммой, она автоматически планируется как задача.

Инструкция:

task = asyncio.create_task(something())
res = await shield(task)

эквивалентна следующей:

res = await something()

за исключением того, что при отмене содержащей её сопрограммы задача, выполняющаяся в something(), не отменяется. С точки зрения something() отмены не произошло. Однако её вызывающий код всё равно отменяется, поэтому выражение «await» по-прежнему вызывает исключение CancelledError.

Если something() отменяется другим способом (то есть изнутри), это также отменит shield().

Если требуется полностью игнорировать отмену (что не рекомендуется), функцию shield() следует использовать вместе с конструкцией try/except, как показано ниже:

task = asyncio.create_task(something())
try:
    res = await shield(task)
except CancelledError:
    res = None

Важно

Сохраните ссылку на задачи, переданные этой функции, чтобы задача не исчезла во время выполнения. Цикл событий хранит только слабые ссылки на задачи. Задача, на которую больше нигде нет ссылок, может быть удалена сборщиком мусора в любой момент, даже до завершения.

Изменено в версии 3.10: Удалён параметр loop.

Устарело с версии 3.10: Выдаётся предупреждение об устаревании, если aw не является объектом, подобным Future, и при этом нет работающего цикла событий.

Тайм-ауты

asyncio.timeout(delay)

Возвращает асинхронный менеджер контекста, который можно использовать для ограничения времени ожидания.

delay может быть равен None или представлять собой число секунд ожидания типа float/int. Если delay равен None, ограничение по времени применяться не будет; это может быть полезно, если при создании менеджера контекста задержка неизвестна.

В обоих случаях после создания менеджера контекста его можно переназначить с помощью Timeout.reschedule().

Пример:

async def main():
    async with asyncio.timeout(10):
        await long_running_task()

Если выполнение long_running_task занимает более 10 секунд, менеджер контекста отменит текущую задачу и обработает возникшее исключение asyncio.CancelledError внутри себя, преобразовав его в TimeoutError, которое можно перехватить и обработать.

Примечание

Именно менеджер контекста asyncio.timeout() преобразует asyncio.CancelledError в TimeoutError, поэтому исключение TimeoutError можно перехватить только за пределами менеджера контекста.

Пример перехвата TimeoutError:

async def main():
    try:
        async with asyncio.timeout(10):
            await long_running_task()
    except TimeoutError:
        print("The long operation timed out, but we've handled it.")

    print("This statement will run regardless.")

Менеджер контекста, созданный с помощью asyncio.timeout(), можно переназначить на другой срок и проверить.

class asyncio.Timeout(when)

Асинхронный менеджер контекста для отмены просроченных корутин.

when должно быть абсолютным временем, когда должен истечь срок действия контекста, согласно часам цикла событий:

• Если when равно None, тайм-аут никогда не сработает.
• Если when < loop.time(), тайм-аут сработает на следующей итерации цикла событий.

when() → float | None

Возвращает текущий срок, или None, если текущий срок не задан.

reschedule(when: float | None)

Переназначает тайм-аут.

expired() → bool

Возвращает значение, указывающее, истёк ли срок действия менеджера контекста.

Пример:

async def main():
    try:
        # We do not know the timeout when starting, so we pass ``None``.
        async with asyncio.timeout(None) as cm:
            # We know the timeout now, so we reschedule it.
            new_deadline = get_running_loop().time() + 10
            cm.reschedule(new_deadline)

            await long_running_task()
    except TimeoutError:
        pass

    if cm.expired():
        print("Looks like we haven't finished on time.")

Менеджеры контекста тайм-аута можно безопасно вкладывать друг в друга.

Добавлено в версии 3.11.

asyncio.timeout_at(when)

Аналогично asyncio.timeout(), но when задаёт абсолютное время окончания ожидания или None.

Пример:

async def main():
    loop = get_running_loop()
    deadline = loop.time() + 20
    try:
        async with asyncio.timeout_at(deadline):
            await long_running_task()
    except TimeoutError:
        print("The long operation timed out, but we've handled it.")

    print("This statement will run regardless.")

Добавлено в версии 3.11.

async asyncio.wait_for(aw, timeout)

Ожидает завершения объекта, допускающего await aw с ограничением по времени.

Если aw — корутина, она автоматически планируется как задача.

timeout может быть равен None или представлять собой число секунд ожидания типа float или int. Если timeout равен None, ожидание продолжается до завершения future.

Если срабатывает тайм-аут, задача отменяется и вызывается исключение TimeoutError.

Чтобы избежать cancellation задачи, оберните её в shield().

Функция будет ждать фактической отмены future, поэтому общее время ожидания может превысить значение timeout. Если во время отмены возникает исключение, оно передаётся дальше.

Если ожидание отменяется, future aw также отменяется.

Пример:

async def eternity():
    # Sleep for one hour
    await asyncio.sleep(3600)
    print('yay!')

async def main():
    # Wait for at most 1 second
    try:
        await asyncio.wait_for(eternity(), timeout=1.0)
    except TimeoutError:
        print('timeout!')

asyncio.run(main())

# Expected output:
#
#     timeout!

Изменено в версии 3.7: Если aw отменяется из-за тайм-аута, wait_for ожидает отмены aw. Ранее исключение TimeoutError вызывалось немедленно.

Изменено в версии 3.10: Параметр loop удалён.

Изменено в версии 3.11: Вызывается TimeoutError вместо asyncio.TimeoutError.

Примитивы ожидания

async asyncio.wait(aws, *, timeout=None, return_when=ALL_COMPLETED)

Запускает экземпляры Future и Task из итерируемого объекта aws параллельно и блокирует выполнение до наступления условия, заданного параметром return_when.

Итерируемый объект aws не должен быть пустым.

Возвращает два набора задач/Future: (done, pending).

Использование:

done, pending = await asyncio.wait(aws)

Параметр timeout (число типа float или int), если он задан, позволяет ограничить максимальное число секунд ожидания до возврата.

Обратите внимание: эта функция не вызывает исключение TimeoutError. Future или задачи, не завершившиеся к моменту срабатывания тайм-аута, просто возвращаются во втором наборе.

Параметр return_when указывает, когда функция должна вернуть результат. Он должен иметь одно из следующих значений:

Константа	Описание
asyncio.FIRST_COMPLETED	Функция вернёт результат, когда любой future завершится или будет отменён.
asyncio.FIRST_EXCEPTION	Функция вернёт результат, когда завершится любой future, вызвав исключение. Если ни один future не вызывает исключения, это значение эквивалентно ALL_COMPLETED.
asyncio.ALL_COMPLETED	Функция вернёт результат, когда все future завершатся или будут отменены.

В отличие от wait_for(), wait() не отменяет future при срабатывании тайм-аута.

Изменено в версии 3.10: Параметр loop удалён.

Изменено в версии 3.11: Передавать объекты-корутины непосредственно в wait() запрещено.

Изменено в версии 3.12: Добавлена поддержка генераторов, выдающих задачи.

asyncio.as_completed(aws, *, timeout=None)

Запускает объекты, допускающие await из итерируемого объекта aws параллельно. Возвращает итератор корутин. Каждую возвращённую корутину можно ожидать, чтобы получить ближайший следующий результат из оставшихся объектов, допускающих await.

Вызывает исключение TimeoutError, если тайм-аут срабатывает до завершения всех Future.

Пример:

for coro in as_completed(aws):
    earliest_result = await coro
    # ...

Изменено в версии 3.10: Параметр loop удалён.

Устарело с версии 3.10: Выдаётся предупреждение об устаревании, если не все объекты, допускающие await, в итерируемом объекте aws подобны Future и при этом цикл событий не запущен.

Изменено в версии 3.12: Добавлена поддержка генераторов, выдающих задачи.

Выполнение в потоках

async asyncio.to_thread(func, /, *args, **kwargs)

Асинхронно выполняет функцию func в отдельном потоке.

Все переданные этой функции аргументы *args и **kwargs напрямую передаются в func. Кроме того, передаётся текущий contextvars.Context, что позволяет обращаться к контекстным переменным из потока цикла событий в отдельном потоке.

Возвращает корутину, которую можно ожидать, чтобы получить результат выполнения func.

Эта функция-корутина предназначена в первую очередь для выполнения функций и методов с привязкой к вводу-выводу, которые в противном случае блокировали бы цикл событий при выполнении в главном потоке. Например:

def blocking_io():
    print(f"start blocking_io at {time.strftime('%X')}")
    # Note that time.sleep() can be replaced with any blocking
    # IO-bound operation, such as file operations.
    time.sleep(1)
    print(f"blocking_io complete at {time.strftime('%X')}")

async def main():
    print(f"started main at {time.strftime('%X')}")

    await asyncio.gather(
        asyncio.to_thread(blocking_io),
        asyncio.sleep(1))

    print(f"finished main at {time.strftime('%X')}")


asyncio.run(main())

# Expected output:
#
# started main at 19:50:53
# start blocking_io at 19:50:53
# blocking_io complete at 19:50:54
# finished main at 19:50:54

Непосредственный вызов blocking_io() в любой корутине блокировал бы цикл событий на всё время выполнения, добавив ещё 1 секунду к общему времени выполнения. Используя asyncio.to_thread(), мы можем вместо этого запустить функцию в отдельном потоке, не блокируя цикл событий.

Примечание

Из-за GIL asyncio.to_thread() обычно можно использовать только для неблокирующего выполнения функций с привязкой к вводу-выводу. Однако в модулях расширения, которые освобождают GIL, или в альтернативных реализациях Python, где его нет, asyncio.to_thread() можно также использовать для функций с привязкой к процессору.

Добавлено в версии 3.9.

Планирование из других потоков

asyncio.run_coroutine_threadsafe(coro, loop)

Передаёт корутину указанному циклу событий. Потокобезопасна.

Возвращает concurrent.futures.Future, с помощью которого можно ожидать результат из другого потока ОС.

Эта функция предназначена для вызова из потока ОС, отличного от того, в котором работает цикл событий. Пример:

# Create a coroutine
coro = asyncio.sleep(1, result=3)

# Submit the coroutine to a given loop
future = asyncio.run_coroutine_threadsafe(coro, loop)

# Wait for the result with an optional timeout argument
assert future.result(timeout) == 3

Если в корутине возникает исключение, возвращённый Future будет уведомлён. Его также можно использовать для отмены задачи в цикле событий:

try:
    result = future.result(timeout)
except TimeoutError:
    print('The coroutine took too long, cancelling the task...')
    future.cancel()
except Exception as exc:
    print(f'The coroutine raised an exception: {exc!r}')
else:
    print(f'The coroutine returned: {result!r}')

См. раздел документации конкурентность и многопоточность.

В отличие от других функций asyncio, эта функция требует явной передачи аргумента loop.

Добавлено в версии 3.5.1.

Интроспекция

asyncio.current_task(loop=None)

Возвращает экземпляр выполняющейся в данный момент задачи Task или None, если ни одна задача не выполняется.

Если loop равен None, для получения текущего цикла используется get_running_loop().

Добавлено в версии 3.7.

asyncio.all_tasks(loop=None)

Возвращает набор ещё не завершённых объектов Task, запущенных циклом.

Если loop равен None, для получения текущего цикла используется get_running_loop().

Добавлено в версии 3.7.

asyncio.iscoroutine(obj)

Возвращает True, если obj является объектом-корутиной.

Добавлено в версии 3.4.

Объект Task

class asyncio.Task(coro, *, loop=None, name=None, context=None, eager_start=False)

Объект Future-like, который запускает Python-сопрограмму. Не является потокобезопасным.

Задачи используются для запуска сопрограмм в циклах событий. Если сопрограмма ожидает объект Future, задача приостанавливает выполнение сопрограммы и ожидает завершения объекта Future. Когда объект Future завершён, выполнение обёрнутой сопрограммы возобновляется.

Циклы событий используют кооперативное планирование: цикл событий выполняет одну задачу за раз. Пока задача ожидает завершения объекта Future, цикл событий выполняет другие задачи и обратные вызовы или выполняет операции ввода-вывода.

Используйте высокоуровневую функцию asyncio.create_task() для создания задач или низкоуровневые функции loop.create_task() или ensure_future(). Создавать экземпляры Task вручную не рекомендуется.

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

Метод cancelled() можно использовать, чтобы проверить, была ли задача отменена. Метод возвращает True, если обёрнутая сопрограмма не подавила исключение CancelledError и действительно была отменена.

asyncio.Task наследует все API класса Future, кроме Future.set_result() и Future.set_exception().

Необязательный именованный аргумент context позволяет указать пользовательский объект contextvars.Context, в котором будет выполняться coro. Если context не указан, задача копирует текущий контекст и впоследствии выполняет свою сопрограмму в скопированном контексте.

Необязательный именованный аргумент eager_start позволяет сразу начать выполнение asyncio.Task при создании задачи. Если задано значение True и цикл событий запущен, задача немедленно начнёт выполнять сопрограмму и продолжит до момента, когда сопрограмма впервые заблокируется. Если сопрограмма вернёт значение или вызовет исключение, не заблокировавшись, задача завершится сразу, не попадая в очередь цикла событий.

Изменено в версии 3.7: Добавлена поддержка модуля contextvars.

Изменено в версии 3.8: Добавлен параметр name.

Устарело начиная с версии 3.10: Выдаётся предупреждение об устаревании, если loop не указан и нет запущенного цикла событий.

Изменено в версии 3.11: Добавлен параметр context.

Изменено в версии 3.12: Добавлен параметр eager_start.

done()

Возвращает True, если задача завершена.

Задача считается завершённой, если обёрнутая сопрограмма вернула значение, вызвала исключение или задача была отменена.

result()

Возвращает результат задачи.

Если задача завершена, возвращается результат обёрнутой сопрограммы (или повторно вызывается исключение, если сопрограмма вызвала его).

Если задача была отменена, этот метод вызывает исключение CancelledError.

Если результат задачи ещё недоступен, этот метод вызывает исключение InvalidStateError.

exception()

Возвращает исключение задачи.

Если обёрнутая сопрограмма вызвала исключение, возвращается это исключение. Если обёрнутая сопрограмма завершилась без ошибок, этот метод возвращает None.

Если задача была отменена, этот метод вызывает исключение CancelledError.

Если задача ещё не завершена, этот метод вызывает исключение InvalidStateError.

add_done_callback(callback, *, context=None)

Добавляет обратный вызов, который будет выполнен, когда задача завершится.

Этот метод следует использовать только в низкоуровневом коде, основанном на обратных вызовах.

Подробности см. в документации метода Future.add_done_callback().

remove_done_callback(callback)

Удаляет callback из списка обратных вызовов.

Этот метод следует использовать только в низкоуровневом коде, основанном на обратных вызовах.

Подробности см. в документации метода Future.remove_done_callback().

get_stack(*, limit=None)

Возвращает список кадров стека для этой задачи.

Если обёрнутая сопрограмма не завершена, возвращается стек в месте её приостановки. Если сопрограмма успешно завершилась или была отменена, возвращается пустой список. Если сопрограмма завершилась из-за исключения, возвращается список кадров трассировки стека.

Кадры всегда упорядочены от самого старого к самому новому.

Для приостановленной сопрограммы возвращается только один кадр стека.

Необязательный аргумент limit задаёт максимальное количество возвращаемых кадров; по умолчанию возвращаются все доступные кадры. Порядок элементов в возвращаемом списке зависит от того, возвращается ли стек или трассировка: для стека возвращаются самые новые кадры, а для трассировки — самые старые. (Это соответствует поведению модуля traceback.)

print_stack(*, limit=None, file=None)

Выводит стек или трассировку для этой задачи.

Для кадров, полученных методом get_stack(), выводятся данные, аналогичные тем, которые предоставляет модуль traceback.

Аргумент limit напрямую передаётся методу get_stack().

Аргумент file — это поток ввода-вывода, в который записывается вывод; по умолчанию вывод записывается в sys.stdout.

get_coro()

Возвращает объект сопрограммы, обёрнутый в Task.

Примечание

Для задач, уже завершившихся сразу, будет возвращено None. См. фабрику задач с немедленным запуском.

Добавлено в версии 3.8.

Изменено в версии 3.12: Из-за добавления немедленного выполнения задач результат может быть равен None.

get_context()

Возвращает объект contextvars.Context, связанный с задачей.

Добавлено в версии 3.12.

get_name()

Возвращает имя задачи.

Если задаче явно не назначено имя, стандартная реализация задачи asyncio генерирует имя по умолчанию при создании экземпляра.

Добавлено в версии 3.8.

set_name(value)

Задаёт имя задачи.

Аргумент value может быть любым объектом, который затем преобразуется в строку.

В стандартной реализации задачи имя отображается в выводе repr() для объекта задачи.

Добавлено в версии 3.8.

cancel(msg=None)

Запрашивает отмену задачи.

Это приводит к тому, что при следующем цикле цикла событий в обёрнутую сопрограмму будет выброшено исключение CancelledError.

Затем сопрограмма может выполнить очистку или даже отклонить запрос, подавив исключение с помощью блока try … … except CancelledError … finally. Поэтому, в отличие от Future.cancel(), метод Task.cancel() не гарантирует отмену задачи, хотя полное подавление отмены встречается редко и настоятельно не рекомендуется. Если сопрограмма всё же решит подавить отмену, ей нужно вызвать Task.uncancel() помимо перехвата исключения.

Изменено в версии 3.9: Добавлен параметр msg.

Изменено в версии 3.11: Параметр msg передаётся от отменённой задачи ожидающей её сопрограмме.

В следующем примере показано, как сопрограммы могут перехватывать запрос на отмену:

async def cancel_me():
    print('cancel_me(): before sleep')

    try:
        # Wait for 1 hour
        await asyncio.sleep(3600)
    except asyncio.CancelledError:
        print('cancel_me(): cancel sleep')
        raise
    finally:
        print('cancel_me(): after sleep')

async def main():
    # Create a "cancel_me" Task
    task = asyncio.create_task(cancel_me())

    # Wait for 1 second
    await asyncio.sleep(1)

    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("main(): cancel_me is cancelled now")

asyncio.run(main())

# Expected output:
#
#     cancel_me(): before sleep
#     cancel_me(): cancel sleep
#     cancel_me(): after sleep
#     main(): cancel_me is cancelled now

cancelled()

Возвращает True, если задача отменена.

Задача считается отменённой, если отмена была запрошена с помощью cancel() и обёрнутая сопрограмма передала дальше выброшенное в неё исключение CancelledError.

uncancel()

Уменьшает счётчик запросов на отмену этой задачи.

Возвращает оставшееся количество запросов на отмену.

Обратите внимание: после завершения выполнения отменённой задачи последующие вызовы uncancel() не оказывают эффекта.

Добавлено в версии 3.11.

Этот метод используется внутренними механизмами asyncio и не предназначен для кода конечных пользователей. В частности, успешная отмена состояния задачи позволяет элементам структурированной конкурентности, таким как группы задач и asyncio.timeout(), продолжить выполнение, ограничив действие отмены соответствующим структурным блоком. Например:

async def make_request_with_timeout():
    try:
        async with asyncio.timeout(1):
            # Structured block affected by the timeout:
            await make_request()
            await make_another_request()
    except TimeoutError:
        log("There was a timeout")
    # Outer code not affected by the timeout:
    await unrelated_code()

Хотя блок с make_request() и make_another_request() может быть отменён из-за тайм-аута, unrelated_code() должен продолжить выполнение даже при наступлении тайм-аута. Это реализовано с помощью uncancel(). Менеджеры контекста TaskGroup используют uncancel() аналогичным образом.

Если по какой-либо причине код конечного пользователя подавляет отмену, перехватывая CancelledError, ему необходимо вызвать этот метод, чтобы сбросить состояние отмены.

cancelling()

Возвращает количество ожидающих запросов на отмену этой задачи, то есть количество вызовов cancel() за вычетом количества вызовов uncancel().

Обратите внимание: если это число больше нуля, но задача всё ещё выполняется, cancelled() всё равно вернёт False. Это связано с тем, что число можно уменьшить, вызвав uncancel(); если количество запросов на отмену станет равным нулю, задачу в итоге можно будет не отменять.

Этот метод используется внутренними механизмами asyncio и не предназначен для кода конечных пользователей. Подробности см. в описании uncancel().

Добавлено в версии 3.11.