python asyncio add tasks dynamically

The main problem with your code was that you called input() directly in your async function. input itself is a blocking function and does not return until a newline or end-of-file is read. This is a problem because Python asynchronous code is still single-threaded, and if there is a blocking function, nothing else will execute. You need to use run_in_executor in this case.

Another problem with your code, although not directly relevant to your question, was that you mixed the pre-python3.7 way of invoking an event loop and the python3.7+ way. Per documentation, asyncio.run is used on its own. If you want to use the pre 3.7 way of invoking a loop, the correct way is

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

or

loop = asyncio.get_event_loop()
asyncio.ensure_future(main())
loop.run_forever()

Since you have a while True in your main(), there's no difference between run_until_complete and run_forever.

Lastly, there is no point in using ensure_future() in your main(). The point of ensure_future is providing a "normal" (i.e. non-async) function a way to schedule things into the event loop, since they can't use the await keyword. Another reason to use ensure_future is if you want to schedule many tasks with high io-bounds (ex. network requests) without waiting for their results. Since you are awaiting the function call, there is naturally no point of using ensure_future.

Here's the modified version:

import asyncio
import os

async def action():
    loop = asyncio.get_running_loop()
    inp = await loop.run_in_executor(None, input, 'Enter a number: ')
    await asyncio.sleep(int(inp))
    os.system(f"say '{inp} seconds waited'")


async def main():
    while True:
        await action()


asyncio.run(main())

In this version, before a user-input is entered, the code execution is alternating between await action() and await loop.run_in_executor(). When no other tasks are scheduled, the event-loop is mostly idle. However, when there are things scheduled (simulated using await sleep()), then the control will be naturally transferred to the long-running task that is scheduled.

One key to Python async programming is you have to ensure the control is transferred back to the event-loop once in a while so other scheduled things can be run. This happens whenever an await is encountered. In your original code, the interpreter get stuck at input() and never had a chance to go back to the event-loop, which is why no other scheduled tasks ever get executed until a user-input is provided.

You can try something like this:

import asyncio


WORKERS = 10


async def worker(q):
    while True:
        t = await q.get()
        await asyncio.sleep(t)
        q.task_done()

        print(f"say '{t} seconds waited'")


async def main():
    q = asyncio.Queue()

    tasks = []

    for _ in range(WORKERS):
        tasks.append(asyncio.create_task(worker(q)))

    print(f'Keep inserting numbers, "q" to quit...')

    while (number := await asyncio.to_thread(input)) != "q":
        q.put_nowait(int(number))

    await q.join()

    for task in tasks:
        task.cancel()

    await asyncio.gather(*tasks, return_exceptions=True)


if __name__ == "__main__":
    asyncio.run(main())

Test:

$ python test.py
Keep inserting numbers, "q" to quit...
1
say '1 seconds waited'
3
2
1
say '1 seconds waited'
say '2 seconds waited'
say '3 seconds waited'
q

Note: Python 3.9+ required due to some new syntax (:=) and function (asyncio.to_thread) in use.

import asyncio

async def aworker(q):
    ''' Worker that takes numbers from the queue and prints them '''    
    while True:
        t = await q.get() # Wait for a number to be put in the queue
        print(f"{t} received {asyncio.current_task().get_coro().__name__}:{asyncio.current_task().get_name()}")
        await asyncio.sleep(t)
        q.task_done()
        print(f"waited for {t} seconds in {asyncio.current_task().get_coro().__name__}:{asyncio.current_task().get_name()}")

async def looper():
    ''' Infinite loop that prints the current task name '''
    i = 0
    while True:
        i+=1
        await asyncio.sleep(1)
        print(f"{i} {asyncio.current_task().get_name()}")
        names = []
        for task in asyncio.all_tasks():
            names.append(task.get_name())
        print(names)

async def main():
    q = asyncio.Queue()
    tasks = []
    
    # create two worker tasks and one infinitely looping task
    tasks.append(asyncio.create_task(aworker(q), name="aworker 1")) # Create a worker which handles input from the queue
    tasks.append(asyncio.create_task(aworker(q), name="aworker 2")) # Create another worker which handles input from the queue
    tasks.append(asyncio.create_task(looper(),name="looper")) # Create a looper task which prints the current task name and the other running tasks
    
    for task in tasks:
        # print the task names thus far
        print(task.get_name())

    print(f'Keep inserting numbers, "q" to quit...')
    
    ''' asyncio.thread names itself Task-1 '''
    while (number := await asyncio.to_thread(input)) != "q":
        try:
            q.put_nowait(int(number))
        except ValueError:
            print("Invalid number")

    await q.join()

    for task in tasks:
        task.cancel()

    await asyncio.gather(*tasks, return_exceptions=True)

if __name__ == "__main__":
    
    asyncio.run(main())