[FIXED] Python asyncio, aiohttp: dynamically updating tasks loop should work until particular condition

Issue

I have simple script for work with api. I create number_of_user users. I go thru loop, create random user create_random_user(), for every user I create task and append task to loop. Task for user creation async def fetch_user_create after getting response create another task for user log in async def fetch_user_log and add it to all tasks. My question: how can I wait until, for example, len(tasks_user) == 2 * number_of_user.

I tried to place await asyncio.sleep(1) - it's work but depends of number_of_user.

Target: wait until condition Is it possible? Or what I did wrong?

async def fetch_user_log(session, data_user):
    async with session.post(url_user_login, data=data_user) as response:

async def fetch_user_create(session, data_user, tasks_user_create):
    async with session.post(url_user_create, data=data_user) as response:
        task2 = asyncio.ensure_future(fetch_user_log(session, data_user))
        tasks_user.append(task2)
        #await asyncio.sleep(1)
        await response.read()
        
        #await asyncio.gather(*tasks_user) - tried

async def run():
    tasks_user = []
    async with ClientSession() as session:
        for i in range(number_of_user):
            data_user = create_random_user()
            task = asyncio.ensure_future(fetch_user_create(session, data_user, tasks_user_create))
            tasks_user.append(task)
        await asyncio.wait(tasks_user)
        #await asyncio.gather(*tasks_user) - tried

loop = asyncio.get_event_loop()
future = asyncio.ensure_future(run())
loop.run_until_complete(future)

Solution

wait until condition can be implemented with asyncio.Condition:

import asyncio

async def test_wait_for(cond, tasks):
    print(".")
    async with cond:
        await cond.wait_for(lambda: (len(tasks)>3))
    print("!")

async def test_add_task(cond, tasks):
    for i in range(6):
        print(i)
        await asyncio.sleep(1)
        async with cond:
            tasks.append(i)
            cond.notify_all()

async def run():
    cond = asyncio.Condition()
    tasks = []
    asyncio.ensure_future(test_wait_for(cond, tasks))
    await test_add_task(cond, tasks)
    


loop = asyncio.get_event_loop()
future = asyncio.ensure_future(run())
loop.run_until_complete(future)

Condition is Event+Lock. Event sent with notify_all unlocks all wait and wait_for coroutines. Lock here also locks tasks array.

Also it can be implemented with asyncio.Event if you move condition to task that send notify.

Answered By - eri