Python 3.11+Асинхронное программированиепонятие

asyncio.Runner

event loop

Контекстный менеджер, который держит один цикл событий для нескольких вызовов run(). То же, что asyncio.run(), но цикл не закрывается между вызовами.

Синтаксис

with asyncio.Runner(*, debug=None, loop_factory=None) as runner: runner.run(coro)

asyncio.run() каждый раз создаёт и закрывает новый цикл, поэтому объекты, привязанные к циклу (соединение, asyncio.Lock, в котором кто-то ждал), во втором вызове ломаются с RuntimeError. Runner решает это: в одном блоке with можно сделать runner.run(a()) и runner.run(b()) на общем цикле — удобно в тестах и в синхронном коде, который зовёт асинхронный по частям. Внутри уже работающего цикла runner.run() вызвать нельзя — там пишут await.

Примеры

import asyncio
async def step(n):
    await asyncio.sleep(0.01)
    return id(asyncio.get_running_loop())
with asyncio.Runner() as runner:
    first = runner.run(step(1))
    second = runner.run(step(2))
    print(first == second)  # → True — цикл один и тот же

import asyncio
with asyncio.Runner() as runner:
    runner.run(asyncio.sleep(0.01))
    loop = runner.get_loop()
    print(loop.is_closed())  # → False
print(loop.is_closed())  # → True — выход из with закрыл цикл

См. также

Открыть в полном глоссарииdocs.python.orgСообщить о неточности