Сообщить об ошибке.

Планирование обратных вызовов низкоуровневым API asyncio в Python

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

Прежде чем что-то делать с циклом событий, его необходимо создать или получить функциями, описанными в разделе "Создание, запуск и получение цикла событий".

Содержание:

Планирование обратных вызовов

loop.call_soon(callback, *args, context=None):

Метод loop.call_soon() планирует обратный вызов callback с аргументами args на следующей итерации цикла событий.

Обратные вызовы вызываются в том порядке, в котором они зарегистрированы. Каждый обратный вызов будет вызываться ровно один раз.

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

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

Этот метод не является потокобезопасным.

loop.call_soon_threadsafe(callback, *args, context=None):

Метод loop.call_soon_threadsafe() потокобезопасный вариант метода loop.call_soon(). Должен использоваться для планирования обратных вызовов из другого потока.

Подробнее смотрите в разделе "Параллелизм и многопоточность в модуле asyncio".

Изменено в Python 3.7: добавлен ключевой аргумент context.

Примечание. Большинство функций планирования asyncio не позволяют передавать ключевые аргументы. Что бы иметь такую возможность используйте функцию functools.partial():

# планирует вызов `print('Hello', flush=True)`
loop.call_soon(
    functools.partial(print, "Hello", flush=True))

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

Планирование отложенных обратных вызовов

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

loop.call_later(delay, callback, *args, context=None):

Метод loop.call_later() планирует обратный вызов callback, который будет вызываться после заданного количества секунд задержки delay Аргумент delay может быть либо int, либо float.

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

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

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

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

Изменено в Python 3.7: добавлен параметр context.

Изменено в Python 3.8: в Python 3.7 и ранее с реализацией цикла событий по умолчанию задержка delay не могла превышать одного дня. Это было исправлено в Python 3.8.

loop.call_at(when, callback, *args, context=None):

Метод loop.call_at() планирует обратный вызов callback, который будет вызываться в заданную абсолютную временную метку when, используя ту же ссылку времени, что и метод loop.time().

Аргумент when может быть либо int, либо float. Поведение этого метода такое же, как и `loop.call_later().

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

Изменено в Python 3.7: добавлен параметр context.

Изменено в Python 3.8: в Python 3.7 и ранее с реализацией цикла событий по умолчанию задержка delay не могла превышать одного дня. Это было исправлено в Python 3.8.

loop.time():

Метод loop.time() возвращает текущее время в виде значения float в соответствии с внутренними монотонными часами цикла событий.

Изменено в Python 3.8: в Python 3.7 и ранее с реализацией цикла событий по умолчанию задержка delay не могла превышать одного дня. Это было исправлено в Python 3.8.

Смотрите также функцию asyncio.sleep().

Дескрипторы обратного вызова.

asyncio.Handle:

Класс asyncio.Handle представляет собой объект-оболочку обратного вызова и имеет методы.

Handle.get_context():

Новое в Python 3.12.

Метод Handle.get_context() возвращает объект contextvars.Context, связанный с дескриптором.

Handle.cancel():

Метод Handle.cancel() отменяет обратный вызов. Если обратный вызов уже был отменен или выполнен, то этот метод не действует.

Handle.cancelled():

Метод Handle.cancel() возвращает True, если обратный вызов был отменен.

Новое в Python 3.7.

asyncio.TimerHandle:

Класс asyncio.TimerHandle является подклассом asyncio.Handle, представляет собой объект-оболочку обратного вызова и имеет один метод.

TimerHandle.when():

Метод TimerHandle.when() возвращает запланированное время обратного вызова в виде секунд float.

Время - это абсолютная временная метка, использующая ту же ссылку на время, что и loop.time().

Новое в Python 3.7.

Пример планирования выполнения функции при помощи обратного вызова.

import asyncio

def mark_done(future, result):
    print(f'Установка будущего результата: {result!r}')
    future.set_result(result)

async def main(loop):
    future = loop.create_future()
    print('Планирование `mark_done()`')
    loop.call_soon(mark_done, future, 'RESULT')
    print('Приостановка работы сопрограммы')
    result = await future
    print(f'Получение результата `future`: {future.result()!r}')
    return result

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    try:
        print('Вход в цикл событий')
        result = loop.run_until_complete(main(loop))
        print(f'Результат: {result!r}')
    finally:
        loop.close()


# Вход в цикл событий
# Планирование `mark_done()`
# Приостановка работы сопрограммы
# Установка будущего результата: 'RESULT'
# Получение результата `future`: 'RESULT'
# Результат: 'RESULT'

Отображение текущего времени с loop.call_later().

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

import asyncio
import datetime

def display_date(end_time, loop):
    print(datetime.datetime.now().strftime('%H:%M:%S'))
    if (loop.time() + 1.0) < end_time:
        loop.call_later(1, display_date, end_time, loop)
    else:
        loop.stop()

loop = asyncio.get_event_loop()

# Планирование первого вызова display_date()
end_time = loop.time() + 5.0
loop.call_soon(display_date, end_time, loop)

# Блокировка вызова прерывается loop.stop()
try:
    loop.run_forever()
finally:
    loop.close()
    
# 08:58:23
# 08:58:24
# 08:58:25
# 08:58:26
# 08:58:27