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

Корутины

Исходный код: 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')}")
  
      # 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 (корутину) в объект 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(). Если группа задач неактивна (например, вход в неё ещё не выполнен, она уже завершена или завершает работу), переданная корутина coro будет закрыта.

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

Пример:

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

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

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.

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

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

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():
    # 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() после перехвата исключения, вызванного одним из объектов, допускающих ожидание, не отменит остальные такие объекты.

Изменено в версии 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)

Ожидает завершения объекта, допускающего ожидание 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)

Одновременно запускает объекты, допускающие ожидание из итерируемого объекта 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 is done. The result can be obtained by
    # awaiting it or calling earliest_connect.result()
    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):
    # next_connect is not one of the original task objects. It must be
    # awaited to obtain the result value or raise the exception of the
    # awaitable that finishes next.
    reader, writer = await next_connect

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

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

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

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

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

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

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() в любой корутине заблокирует цикл событий на время её выполнения, добавив к общему времени ещё одну секунду. Используя asyncio.to_thread(), можно вместо этого запустить её в отдельном потоке, не блокируя цикл событий.

Примечание

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

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

set_name(value)

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

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

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

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

cancel(msg=None)

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

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

Метод обеспечивает вызов исключения 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, ему необходимо вызвать этот метод, чтобы сбросить состояние отмены.

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

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

cancelling()

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

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

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

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