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

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

Корутины
Awaitables
Запуск программы asyncio
Создание задач
Ожидание
Запуск задач одновременно
Защита от отмены
Таймауты
Примитивы ожидания
Планирование из других потоков
Интроспекция
Объект задачи
Корутины на основе генераторов

Корутины

Корутины объявленные с помощью синтаксиса async/await — предпочтительный способ написания приложений asyncio. Например, следующий фрагмент кода (требует Python 3.7+) выводит “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
```

Awaitables

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

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

Корутины

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

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

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

asyncio.run(main())

Важно

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

функция корутины: функция async def;
объект корутины: объект, возвращаемый при вызове функции корутины.

asyncio также поддерживает устаревшие генераторные корутины.

Задачи

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

Когда корутина оборачивается в задачу с помощью функций, таких как 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 — это специальный низкоуровневый awaitable объект, который представляет конечный результат асинхронной операции.

Когда ожидается объект 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().

Запуск программы asyncio

asyncio.run(coro, *, debug=False)

Выполнить корутину coro и вернуть результат.

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

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

Если debug True, цикл событий будет запущен в отладочном режиме.

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

Пример:

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

asyncio.run(main())

Новая в версии 3.7.

Примечание

Исходный код для asyncio.run() можно найти в Lib/asyncio/runners.py.

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

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

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

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

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

Эта функция была добавлена в Python 3.7. До Python 3.7 можно было использовать низкоуровневую функцию asyncio.ensure_future():

async def coro():
    ...

# In Python 3.7+
task = asyncio.create_task(coro())
...

# This works in all Python versions but is less readable
task = asyncio.ensure_future(coro())
...

Новая в версии 3.7.

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

Ожидание

coroutine asyncio.sleep(delay, result=None, *, loop=None)

Ожидание delay секунд.

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

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

Устарело начиная с версии 3.8, будет удалено в версии 3.10: Параметр loop.

Пример корутины, отображающей текущую дату каждую секунду в течение 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())

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

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

Запустить объекты-ожидания в последовательности aws параллельно.

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

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

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

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

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

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

Устарело начиная с версии 3.8, будет удалено в версии 3.10: Параметр loop.

Пример:

import asyncio

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

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

asyncio.run(main())

# Expected output:
#
#     Task A: Compute factorial(2)...
#     Task B: Compute factorial(2)...
#     Task C: Compute factorial(2)...
#     Task A: factorial(2) = 2
#     Task B: Compute factorial(3)...
#     Task C: Compute factorial(3)...
#     Task B: factorial(3) = 6
#     Task C: Compute factorial(4)...
#     Task C: factorial(4) = 24

Примечание

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

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

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

awaitable asyncio.shield(aw, *, loop=None)

Защищает объект-ожидание от cancelled.

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

Выражение:

res = await shield(something())

эквивалентно:

res = await something()

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

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

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

try:
    res = await shield(something())
except CancelledError:
    res = None

Устарело начиная с версии 3.8, будет удалено в версии 3.10: Параметр loop.

Таймауты

coroutine asyncio.wait_for(aw, timeout, *, loop=None)

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

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

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

Если таймаут истекает, задача отменяется, и возбуждается asyncio.TimeoutError.

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

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

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

Устарело начиная с версии 3.8, будет удалено в версии 3.10: Параметр loop.

Пример:

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 asyncio.TimeoutError:
        print('timeout!')

asyncio.run(main())

# Expected output:
#
#     timeout!

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

Ожидающие примитивы

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

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

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

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

done, pending = await asyncio.wait(aws)

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

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

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

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

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

Устарело начиная с версии 3.8: Если любой выполнимый объект в aws является корутиной, он автоматически планируется как задача. Передача объектов корутин в wait() напрямую устарела, так как приводит к непонятному поведению.

Устарело начиная с версии 3.8, будет удалено в версии 3.10: Параметр loop.

Примечание

wait() автоматически планирует корутины как задачи и впоследствии возвращает эти неявно созданные объекты задач в (done, pending) наборы. Поэтому следующий код не будет работать как ожидалось:

async def foo():
    return 42

coro = foo()
done, pending = await asyncio.wait({coro})

if coro in done:
    # This branch will never be run!

Вот как можно исправить приведенный выше фрагмент:

async def foo():
    return 42

task = asyncio.create_task(foo())
done, pending = await asyncio.wait({task})

if task in done:
    # Everything will work as expected now.

Устарело начиная с версии 3.8: Передача объектов корутин в wait() напрямую устарела.

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

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

Вызывает asyncio.TimeoutError, если таймаут наступит до того, как все задачи будут выполнены.

Устарело начиная с версии 3.8, будет удалено в версии 3.10: Параметр loop.

Пример:

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

Расписание из других потоков

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

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

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

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

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

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

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

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

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

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

Задачи поддерживают модуль contextvars. При создании задачи она копирует текущий контекст и затем выполняет свою корутину в скопированном контексте.

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

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

Устарело начиная с версии 3.8, будет удалено в версии 3.10: Параметр loop.

cancel()

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

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

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

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

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, брошенное в неё.

done()

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

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

END_OF_DOCUMENT_MARKER

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

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

Новая в версии 3.8.

get_name()

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

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

Новая в версии 3.8.

set_name(value)

Устанавливает имя задачи.

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

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

Новая в версии 3.8.

classmethod all_tasks(loop=None)

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

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

Устарело начиная с версии 3.7, будет удалено в версии 3.9: Не вызывайте этот метод как метод задачи. Используйте функцию asyncio.all_tasks() вместо этого.

classmethod current_task(loop=None)

Возвращает текущую выполняющуюся задачу или None.

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

Устарело начиная с версии 3.7, будет удалено в версии 3.9: Не вызывайте этот метод как метод задачи. Используйте функцию asyncio.current_task() вместо этого.

Генераторные корутины

Примечание

Поддержка генераторных корутин устарела и будет удалена в Python 3.10.

Генераторные корутины предшествуют синтаксису async/await. Это генераторы Python, которые используют yield from выражения для ожидания на Futures и других корутин.

Генераторные корутины должны быть декорированы с помощью @asyncio.coroutine, хотя это не обязательно.

@asyncio.coroutine

Декоратор для маркировки генераторных корутин.

Этот декоратор позволяет корутинам на основе генераторов из прошлого быть совместимыми с кодом async/await:

@asyncio.coroutine
def old_style_coroutine():
    yield from asyncio.sleep(1)

async def main():
    await old_style_coroutine()

Этот декоратор не следует использовать для корутин async def.

Устарело начиная с версии 3.8, будет удалено в версии 3.10: Используйте async def вместо этого.

asyncio.iscoroutine(obj)

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

Этот метод отличается от inspect.iscoroutine(), поскольку возвращает True для генераторных корутин.

asyncio.iscoroutinefunction(func)

Возвращает True если func является функцией корутины.

Этот метод отличается от inspect.iscoroutinefunction(), поскольку возвращает True для функций генераторных корутин, декорированных @coroutine.

© 2001–2022 Python Software Foundation
Licensed under the PSF License.
https://docs.python.org/3.8/library/asyncio-task.html