Функции создания и получения текущего цикла событий

Цикл событий - это ядро любого приложения на основе модуля asyncio. Объект цикла событий, своими методами запускает асинхронные задачи и обратные вызовы, выполняют сетевые операции ввода-вывода, а так же запускают подпроцессы.

Разработчики приложений, для запуска цикла событий асинхронного кода, должны использовать высокоуровневые функции модуля asyncio, такие как asyncio.run(), так как редко нужно ссылаться на объект цикла или вызывать его методы. Низкоуровневые функции и объекты (низкоуровневый API) предназначен для авторов библиотек и фреймворков, которым требуется более тонкий контроль над поведением цикла событий.

Функции получения, установки и создания цикла событий.

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

asyncio.get_running_loop() - получает запущенный цикл событий,
asyncio.get_event_loop() - получить текущий цикл событий,
asyncio.set_event_loop() - устанавливает цикл loop как текущий цикл событий,
asyncio.new_event_loop() - создает новый объект цикла событий,
Возможности низкоуровнего API цикла событий,
Реализации цикла событий,
Пример использования низкоуровнего API циклов событий.

`asyncio.get_running_loop()`:

Функция asyncio.get_running_loop() возвращает запущенный цикл событий в текущем потоке ОС.

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

`asyncio.get_event_loop()`:

Функция asyncio.get_event_loop() получает текущий цикл событий.

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

Если не установлен запущенный цикл событий, функция вернет результат вызова get_event_loop_policy().get_event_loop().

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

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

С версии Python 3.12: если текущий цикл событий отсутствует, то выдается предупреждение об устаревании. В будущих версиях Python это станет ошибкой.

`asyncio.set_event_loop(loop)`:

Функция asyncio.set_event_loop() устанавливает цикл loop как текущий цикл событий для текущего потока ОС.

`asyncio.new_event_loop()`:

Метод asyncio.new_event_loop() создает новый объект цикла событий loop.

Возможности низкоуровнего API цикла событий:

Реализации цикла событий.

Asyncio поставляется с двумя разными реализациями цикла событий: SelectorEventLoop и ProactorEventLoop.

По умолчанию asyncio настроен на использование SelectorEventLoop в Unix и ProactorEventLoop в Windows.

`asyncio.SelectorEventLoop`:

Класс asyncio.SelectorEventLoop представляет собой цикл событий, основанный на модуле selectors.

Использует наиболее эффективный селектор, доступный для данной платформы. Также можно вручную настроить точную реализацию селектора, который будет использоваться:

import asyncio
import selectors

selector = selectors.SelectSelector()
loop = asyncio.SelectorEventLoop(selector)
asyncio.set_event_loop(loop)

Доступность: Unix, Windows.

`asyncio.ProactorEventLoop`:

Класс asyncio.ProactorEventLoop представляет собой цикл событий для Windows, использующий порты завершения ввода/вывода IOCP.

Доступность: Windows.

Более подробно смотрите документацию MSDN по портам завершения ввода/вывода.

`asyncio.AbstractEventLoop`:

Класс asyncio.AbstractEventLoop представляет собой абстрактный базовый класс для asyncio-совместимых циклов событий.

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

Примеры использования низкоуровнего API циклов событий:

Обратите внимание, что пример целенаправленно показывают, как использовать низкоуровневые API-интерфейсы цикла событий, таких как loop.run_forever() и loop.call_soon(). Современные приложения asyncio очень редко необходимо писать таким образом.

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

Привет, мир с `loop.call_soon()`.

Пример использования метода loop.call_soon() для планирования обратного вызова. Обратный вызов отображает "Hello World", а затем останавливает цикл событий.

import asyncio

def hello_world(loop):
    """
    Обратный вызов для печати "Hello World" 
    и остановки цикла событий
    """
    print('Hello World')
    # прерываем цикл событий
    loop.stop()

# получение текущего цикла событий
loop = asyncio.get_event_loop()

# Планирует вызов функции `hello_world()`
loop.call_soon(hello_world, loop)

# Блокирующий вызов прерывается loop.stop()
try:
    loop.run_forever()
finally:
    # освобождение ресурсов
    # потребляемых циклом событий
    loop.close()

Функции создания и получения текущего цикла событий

Функции получения, установки и создания цикла событий.

asyncio.get_running_loop():

asyncio.get_event_loop():

asyncio.set_event_loop(loop):

asyncio.new_event_loop():