Can asyncio.sleep be stopped?

Question

Can you stop asyncio.sleep() from sleeping? I want to run some code in a coroutine either after waiting 20 seconds or if a Boolean is true, is there any way to do that? I can only think of achieving it like this:

while t < 20 and not value:
    asyncio.sleep(1)
    t +=1
...

Is there a more pythonic way to do this?

Answer 1

IIUC, you can use asyncio.Future() + asyncio.wait_for():

import asyncio


async def task_fn(fut: asyncio.Future):
    await asyncio.sleep(2)  # <-- after 2 seconds set the Future's result to True
    fut.set_result(True)


async def main():
    task = asyncio.Task(task_fn(fut := asyncio.Future()))

    try:
        await asyncio.wait_for(fut, 3)    # <-- wait 3 seconds for the future result
        print("OK, task finished within 3 seconds")
    except TimeoutError:
        print("Error, task didn't finish within 3 seconds")


asyncio.run(main())

This prints:

OK, task finished within 3 seconds

---

If you change the task_fn to:

async def task_fn(fut: asyncio.Future):
    await asyncio.sleep(5)  # <--- notice the 5 seconds
    fut.set_result(True)

Then the result will be:

Error, task didn't finish within 3 seconds