Event Loop

Отримання циклу подій

Створення і отримання циклу подій

• asyncio.get_event_loop(): Повертає поточний цикл подій або створює новий, якщо поточного немає.
• asyncio.new_event_loop(): Створює новий цикл подій.
• asyncio.set_event_loop(loop): Встановлює вказаний цикл подій як поточний.

import asyncio

loop = asyncio.get_event_loop()
print(loop)  # Поточний цикл подій
new_loop = asyncio.new_event_loop()
asyncio.set_event_loop(new_loop)
print(asyncio.get_event_loop())  # Новий встановлений цикл подій

Запуск циклу подій

• run_forever(): Запускає цикл подій і продовжує його виконання до виклику stop().
• run_until_complete(future): Запускає цикл подій і завершує його після завершення заданої корутини або майбутнього об'єкта.

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

loop = asyncio.get_event_loop()
loop.run_until_complete(hello())
loop.close()

import asyncio

async def main():
    await asyncio.sleep(1)
    print("Task completed")

loop = asyncio.get_event_loop()
task = loop.create_task(main())
loop.run_until_complete(task)

Зупинка циклу подій

• stop(): Зупиняє цикл подій.
• is_running(): Повертає True, якщо цикл подій запущений.

import asyncio

loop = asyncio.get_event_loop()
loop.stop()
print(loop.is_running())  # False

Важливі методи циклу подій

Метод call_soon(callback, *args)

Планує виклик функції callback з аргументами *args якомога швидше.

import asyncio

def my_callback():
    print("Callback executed")

loop = asyncio.get_event_loop()
loop.call_soon(my_callback)
loop.run_forever()

Метод call_later(delay, callback, *args)

Планує виклик функції callback з аргументами *args через delay секунд.

import asyncio

def my_callback():
    print("Callback executed after delay")

loop = asyncio.get_event_loop()
loop.call_later(2, my_callback)
loop.run_forever()

Метод call_at(when, callback, *args)

Планує виклик функції callback з аргументами *args в момент часу when.

import asyncio
import time

def my_callback():
    print("Callback executed at specific time")

loop = asyncio.get_event_loop()
when = loop.time() + 2  # Через 2 секунди від поточного часу циклу подій
loop.call_at(when, my_callback)
loop.run_forever()