Coroutines and tasksКорутины и задачи

В этом разделе представлены высокоуровневые API asyncio для работы с корутинами и задачами.

• Корутины

• Ожидаемые объекты

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

• Отмена задачи

• Группы задач

• Ожидание

• Конкурентное выполнение задач

• Нетерпеливая фабрика задач

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

• Тайм-ауты

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

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

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

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

• Объект задачи

CoroutinesКорутины

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

---

Корутины, объявленные с синтаксисом async/await, являются предпочтительным способом написания asyncio-приложений. Например, следующий фрагмент кода выводит “hello”, ждёт 1 секунду и затем выводит “world”:

>>> 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()» (см. пример выше).

• Ожидание корутины. Следующий фрагмент кода выведет «hello» после ожидания в 1 секунду, а затем выведет «world» после ожидания ещё 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')}")
  
      # Дождаться завершения обеих задач (должно занять
      # около 2 секунд.)
      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')}")
  
      # await является неявным при выходе из контекстного менеджера.
  
      print(f"finished at {time.strftime('%X')}")
  

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

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

AwaitablesОжидаемые объекты

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

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

Корутины

Корутины Python являются ожидаемыми объектами и поэтому могут ожидаться из других корутин:

import asyncio

async def nested():
    return 42

async def main():
    # Ничего не происходит, если просто вызвать "nested()".
    # Создаётся объект корутины, но он не ожидается,
    # поэтому корутина *вообще не выполнится*.
    nested()  # вызовет предупреждение "RuntimeWarning".

    # Теперь сделаем по-другому и ожидаем её:
    print(await nested())  # выведет "42".

asyncio.run(main())

Важно

В этой документации термин «корутина» может использоваться для двух тесно связанных понятий:

• корутинная функция: функция async def;

• объект корутины: объект, возвращаемый вызовом корутинной функции.

Задачи

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

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

import asyncio

async def nested():
    return 42

async def main():
    # Запланировать выполнение nested() в ближайшее время конкурентно
    # с "main()".
    task = asyncio.create_task(nested())

    # "task" теперь можно использовать для отмены "nested()" или
    # может быть просто ожидана с помощью await до завершения:
    await task

asyncio.run(main())

Фьючерсы

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

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

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

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

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

async def main():
    await function_that_returns_a_future_object()

    # это также допустимо:
    await asyncio.gather(
        function_that_returns_a_future_object(),
        some_python_coroutine()
    )

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

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

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

---

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

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

Полная сигнатура функции в значительной степени совпадает с сигнатурой конструктора Task (или фабрики) – все ключевые аргументы этой функции передаются этому интерфейсу.

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

Необязательный аргумент, передаваемый только по ключу, eager_start позволяет указать, должна ли задача выполняться немедленно при вызове create_task или быть запланированной на потом. Если eager_start не передан, будет использоваться режим, установленный loop.set_task_factory().

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

Примечание

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

Важно

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

background_tasks = set()

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

    # Добавить задачу в набор. Это создает сильную ссылку.
    background_tasks.add(task)

    # Чтобы предотвратить постоянное хранение ссылок на завершенные задачи,
    # заставьте каждую задачу удалять свою собственную ссылку из набора после
    # завершения:
    task.add_done_callback(background_tasks.discard)

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

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

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

Изменено в версии 3.14: Добавлен параметр eager_start путём передачи всех kwargs.

Task cancellationОтмена задачи

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

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

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

Task groupsГруппы задач

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

class asyncio.TaskGroup¶

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

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

create_task(coro, *, name=None, context=None, eager_start=None, **kwargs)¶

Создаёт задачу в этой группе задач. Сигнатура соответствует asyncio.create_task(). Если группа задач неактивна (например, ещё не вошли, уже завершена или в процессе завершения), закрывает переданную coro.

Изменено в версии 3.13: Закрывает переданную корутину, если группа задач не активна.

Изменено в версии 3.14: Передаёт все kwargs в loop.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, как в предыдущем абзаце.

Группы задач стараются не путать внутреннюю отмену, используемую для «пробуждения» своего __aexit__(), с запросами на отмену задачи, в которой они выполняются, поступающими от других сторон. В частности, когда одна группа задач синтаксически вложена в другую, и обе одновременно сталкиваются с исключением в одной из своих дочерних задач, внутренняя группа задач обрабатывает свои исключения, а затем внешняя группа задач получает ещё одну отмену и обрабатывает свои собственные исключения.

В случае, когда группа задач отменяется извне и также должна возбудить ExceptionGroup, она вызывает метод cancel() родительской задачи. Это гарантирует, что asyncio.CancelledError будет возбуждено при следующем await, поэтому отмена не теряется.

Группы задач сохраняют счётчик отмен, сообщаемый asyncio.Task.cancelling().

Изменено в версии 3.13: Улучшена обработка одновременных внутренних и внешних отмен и корректное сохранение счетчиков отмен.

Terminating a task groupЗавершение группы задач

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

import asyncio
from asyncio import TaskGroup

class TerminateTaskGroup(Exception):
    """Исключение, вызываемое для завершения группы задач."""

async def force_terminate_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:
            # породить несколько задач
            group.create_task(job(1, 0.5))
            group.create_task(job(2, 1.5))
            # спать 1 секунду
            await asyncio.sleep(1)
            # добавить задачу, вызывающую исключение, чтобы принудительно завершить группу
            group.create_task(force_terminate_task_group())
    except* TerminateTaskGroup:
        pass

asyncio.run(main())

Ожидаемый вывод:

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

SleepingСплю

async asyncio.sleep(delay, result=None)¶

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

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

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

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

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

import asyncio
import datetime as dt

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

asyncio.run(display_date())

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

Изменено в версии 3.13: Возбуждает ValueError, если delay равен nan.

Running tasks concurrentlyПараллельное выполнение задач

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

Выполняет ожидаемые объекты из последовательности aws параллельно.

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

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

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

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

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

Если какая-либо задача или 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():
    # Запланировать три вызова *одновременно*:
    L = await asyncio.gather(
        factorial("A", 2),
        factorial("B", 3),
        factorial("C", 4),
    )
    print(L)

asyncio.run(main())

# Ожидаемый вывод:
#
#     Задача A: Вычисление factorial(2), сейчас i=2...
#     Задача B: Вычисление factorial(3), сейчас i=2...
#     Задача C: Вычисление factorial(4), сейчас i=2...
#     Задача A: factorial(2) = 2
#     Задача B: Вычисление factorial(3), сейчас i=3...
#     Задача C: Вычисление factorial(4), сейчас i=3...
#     Задача B: factorial(3) = 6
#     Задача C: Вычисление factorial(4), сейчас i=4...
#     Задача 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-подобными объектами и нет работающего цикла событий.

Eager task factoryФабрика немедленных задач

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.

Shielding from cancellationЗащита от отмены

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, и нет работающего цикла событий.

TimeoutsТайм-ауты

asyncio.timeout(delay)¶

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

delay может быть None или числом с плавающей точкой/целым числом секунд ожидания. Если 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)¶

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

Предпочтительно использовать asyncio.timeout() или asyncio.timeout_at(), а не создавать экземпляр Timeout напрямую.

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

• Если when равно None, тайм-аут никогда не сработает.

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

when() → float | None¶

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

reschedule(when: float | None)¶

Перенести тайм-аут.

expired() → bool¶

Возвращает, превысил ли контекстный менеджер свой срок (истек).

Пример:

async def main():
    try:
        # Мы не знаем тайм-аут при запуске, поэтому передаем ``None``.
        async with asyncio.timeout(None) as cm:
            # Теперь мы знаем тайм-аут, поэтому переназначаем его.
            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)¶

Ожидает завершения aw ожидаемого объекта с тайм-аутом.

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

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

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

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

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

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

Пример:

async def eternity():
    # Ожидание в течение часа
    await asyncio.sleep(3600)
    print('yay!')

async def main():
    # Ждать не более 1 секунды
    try:
        await asyncio.wait_for(eternity(), timeout=1.0)
    except TimeoutError:
        print('timeout!')

asyncio.run(main())

# Ожидаемый вывод:
#
#     Тайм-аут!

Изменено в версии 3.7: Когда aw отменяется из-за тайм-аута, wait_for ожидает отмены aw. Ранее это вызывало TimeoutError немедленно.

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

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

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

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 (число с плавающей точкой или целое) может использоваться для управления максимальным количеством секунд ожидания перед возвратом.

Обратите внимание, что эта функция не вызывает 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)¶

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

Объект, возвращаемый as_completed(), можно итерировать как асинхронный итератор или обычный итератор. При использовании асинхронной итерации исходно переданные ожидаемые объекты возвращаются, если они являются задачами или future'ами. Это упрощает сопоставление ранее запланированных задач с их результатами. Пример:

ipv4_connect = create_task(open_connection("127.0.0.1", 80))
ipv6_connect = create_task(open_connection("::1", 80))
tasks = [ipv4_connect, ipv6_connect]

async for earliest_connect in as_completed(tasks):
    # earliest_connect завершена. Результат можно получить с помощью await или вызова earliest_connect.result()
    # next_connect не является одним из исходных объектов задач. Его необходимо ожидать (await) для получения значения результата или возбуждения исключения ожидаемого объекта, завершающегося следующим.
    reader, writer = await earliest_connect

    if earliest_connect is ipv6_connect:
        print("IPv6 connection established.")
    else:
        print("IPv4 connection established.")

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

При использовании в качестве обычного итератора каждая итерация возвращает новую корутину, которая возвращает результат или возбуждает исключение следующего завершённого ожидаемого объекта. Этот шаблон совместим с версиями Python старше 3.13:

ipv4_connect = create_task(open_connection("127.0.0.1", 80))
ipv6_connect = create_task(open_connection("::1", 80))
tasks = [ipv4_connect, ipv6_connect]

for next_connect in as_completed(tasks):
    # Прямое наследование
    # Дополнительный метод, не требуемый ABC
    # Требуемый абстрактный метод
    reader, writer = await next_connect

Исключение TimeoutError возбуждается, если тайм-аут наступает до того, как все ожидаемые объекты завершены. Оно возбуждается циклом async for во время асинхронной итерации или корутинами, возвращаемыми при обычной итерации.

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

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

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

Изменено в версии 3.13: Результат теперь можно использовать как асинхронный итератор, так и как обычный итератор (ранее он был только обычным итератором).

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

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')}")
    # Обратите внимание, что time.sleep() можно заменить на любую блокирующую
    # операцию, связанную с вводом-выводом, например, файловые операции.
    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())

# Ожидаемый вывод:
#
# запущен main в 19:50:53
# запущен blocking_io в 19:50:53
# blocking_io завершён в 19:50:54
# main завершён в 19:50:54

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

Примечание

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

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

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

asyncio.run_coroutine_threadsafe(coro, loop)¶

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

Возвращает объект concurrent.futures.Future для ожидания результата из другого потока ОС.

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

def in_thread(loop: asyncio.AbstractEventLoop) -> None:
    # Выполнить некоторый блокирующий ввод-вывод
    pathlib.Path("example.txt").write_text("hello world", encoding="utf8")

    # Создать корутину
    coro = asyncio.sleep(1, result=3)

    # Отправить корутину в заданный цикл событий
    future = asyncio.run_coroutine_threadsafe(coro, loop)

    # Ожидать результат с необязательным аргументом тайм-аута
    assert future.result(timeout=2) == 3

async def amain() -> None:
    # Получить работающий цикл событий
    loop = asyncio.get_running_loop()

    # Выполнить что-то в потоке
    await asyncio.to_thread(in_thread, loop)

Также возможно выполнение в обратном направлении. Пример:

@contextlib.contextmanager
def loop_in_thread() -> Generator[asyncio.AbstractEventLoop]:
    loop_fut = concurrent.futures.Future[asyncio.AbstractEventLoop]()
    stop_event = asyncio.Event()

    async def main() -> None:
        loop_fut.set_result(asyncio.get_running_loop())
        await stop_event.wait()

    with concurrent.futures.ThreadPoolExecutor(1) as tpe:
        complete_fut = tpe.submit(asyncio.run, main())
        for fut in concurrent.futures.as_completed((loop_fut, complete_fut)):
            if fut is loop_fut:
                loop = loop_fut.result()
                try:
                    yield loop
                finally:
                    loop.call_soon_threadsafe(stop_event.set)
            else:
                fut.result()

# Создать цикл событий в другом потоке
with loop_in_thread() as loop:
    # Создать корутину
    coro = asyncio.sleep(1, result=3)

    # Отправить корутину в заданный цикл событий
    future = asyncio.run_coroutine_threadsafe(coro, loop)

    # Ожидать результат с необязательным аргументом тайм-аута
    assert future.result(timeout=2) == 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}')

Смотрите раздел concurrency and multithreading документации.

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

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

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

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 objectОбъект задачи

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(). Ручное создание экземпляров задач не рекомендуется.

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

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

asyncio.Task наследует от Future все его API, кроме 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)¶

Удаляет колбэк из списка колбэков.

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

См. документацию Future.remove_done_callback() для получения дополнительных сведений.

get_stack(*, limit=None)¶

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

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

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

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

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

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

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

Вывод аналогичен выводу модуля traceback для фреймов, полученных с помощью get_stack().

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

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

get_coro()¶

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

Примечание

Для задач, которые уже были выполнены с нетерпением (eager), будет возвращено None. См. Eager Task Factory.

Добавлено в версии 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)¶

Запросить отмену задачи.

Если задача уже завершена или отменена, вернуть False, в противном случае вернуть True.

Метод организует выбрасывание исключения CancelledError в обёрнутую корутину на следующем цикле событий.

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

Если отменяемая задача в данный момент ожидает объект, подобный future, этот ожидаемый объект также будет отменён. Эта отмена распространяется вниз по всей цепочке ожидаемых объектов.

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

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

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

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

    try:
        # Ожидать 1 час
        await asyncio.sleep(3600)
    except asyncio.CancelledError:
        print('cancel_me(): cancel sleep')
        raise
    finally:
        print('cancel_me(): after sleep')

async def main():
    # Создать задачу "cancel_me"
    task = asyncio.create_task(cancel_me())

    # Ждать 1 секунду
    await asyncio.sleep(1)

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

asyncio.run(main())

# Ожидаемый вывод:
#
#     cancel_me(): перед сном
#     cancel_me(): отмена сна
#     cancel_me(): после сна
#     main(): cancel_me теперь отменена

cancelled()¶

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

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

uncancel()¶

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

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

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

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

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

async def make_request_with_timeout():
    try:
        async with asyncio.timeout(1):
            # Структурированный блок, затронутый тайм-аутом:
            await make_request()
            await make_another_request()
    except TimeoutError:
        log("There was a timeout")
    # Внешний код, не затронутый тайм-аутом:
    await unrelated_code()

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

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

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

Изменено в версии 3.13: Изменено: отменять ожидающие запросы на отмену при достижении нуля.

cancelling()¶

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

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

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

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