Содержание страницы

Корутины и задачи¶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())

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

asyncio.run(main())

Future

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)¶

Wrap the coro coroutine into a Task and schedule its execution. Return the Task object.

Полная сигнатура функции в основном такая же, как у конструктора (или фабрики) 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¶

An asynchronous context manager holding a group of tasks. Tasks can be added to the group using create_task(). All tasks are awaited when the context manager exits.

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

create_task(coro, *, name=None, context=None, eager_start=None, **kwargs)¶: Создаёт задачу в этой группе задач. Сигнатура совпадает с asyncio.create_task(). Если группа задач неактивна (например, ещё не вошли, уже завершена или находится в процессе остановки), указанная coro будет закрыта.

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

Изменено в версии 3.14: Передаёт все kwargs в loop.create_task()

cancel()¶

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

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

Если cancel() вызывается до входа в группу задач, группа будет отменена при входе. Это полезно для сценариев, когда один фрагмент кода передаёт неиспользуемый экземпляр asyncio.TaskGroup другому, чтобы иметь возможность отменить всё, выполняемое внутри группы.

cancel() идемпотентен и может быть вызван после того, как группа задач уже завершена.

Некоторые способы использования cancel():

вызвать его из тела группы задач на основе какого-либо условия или события
передать экземпляр группы задач дочерним задачам через create_task(), что позволяет дочерней задаче условно отменить всю группу
передать экземпляр группы задач или привязанный метод cancel() другой задаче до открытия группы задач, что позволяет удалённую отмену

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

Пример:

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() в этой корутине). Также есть возможность запросить завершение всей группы задач с помощью tg.cancel(), исходя из некоторого условия. Как только последняя задача завершена и блок 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__() вызывается с установленным исключением), это обрабатывается так же, как если бы одна из задач завершилась ошибкой: оставшиеся задачи отменяются и ожидаются, а исключения, не связанные с отменой, группируются в gруппу исключений и возбуждаются. Исключение, переданное в __aexit__(), если только оно не является asyncio.CancelledError, также включается в группу исключений. Тот же особый случай применяется для KeyboardInterrupt и SystemExit, как в предыдущем абзаце.

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

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

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

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

Ожидание ¶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, ограничение по времени не применяется; это может быть полезно, если delay неизвестен на момент создания менеджера контекста.

В любом случае, менеджер контекста можно перенастроить после создания с помощью 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.12.

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.12.

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¶	Функция вернёт результат, когда все futures завершатся или будут отменены.

В отличие от wait_for(), wait() не отменяет futures при возникновении таймаута.

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

Примечание

Это вернёт 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)¶

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

Если задача уже выполнена (done) или отменена (cancelled), возвращает 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()¶

Return True if the Task is cancelled.

The Task is cancelled when the cancellation was requested with cancel() and the wrapped coroutine propagated the CancelledError exception thrown into it.

uncancel()¶

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

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

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

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

Этот метод используется внутренними механизмами 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.12.

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

Корутины¶Coroutines

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

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

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

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

Ожидание¶Sleeping

Одновременное выполнение задач¶Running tasks concurrently

Фабрика задач с немедленным выполнением¶Eager task factory

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

Тайм-ауты¶Timeouts

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

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

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

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

Объект задачи¶Task object

Корутины ¶Coroutines

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

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

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

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

Ожидание ¶Sleeping

Одновременное выполнение задач ¶Running tasks concurrently

Фабрика задач с немедленным выполнением ¶Eager task factory

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

Тайм-ауты ¶Timeouts

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

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

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

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

Объект задачи ¶Task object