[FIXED] How to properly open and close an event-loop in a for loop?

Issue

I have an asyncio loop watch_book nested in wallet_loop and main I would like to execute in a for loop. What is the proper way to do that ?

My problem is that after the first account terminate it's loop and close it, the second account create a new loop but it still complain that the loop is closed. What is the reason for this ?

async def watch_book(account, client, market):
    while True:
        ob = await client.watch_order_book(market.symbol)
        if stop():
            break
        await client.sleep(1000)

async def wallet_loop(account, loop, wallet):
    client = getattr(name, exid)({'enableRateLimit': True, 'asyncio_loop': loop, })
    do_more_stuff()
    ws_loops = [watch_book(account, client, market) for market in markets]
    await asyncio.gather(*ws_loops)
    await client.close()


async def main(account, loop, wallets):
    wallet_loops = [wallet_loop(account, loop, wallet) for wallet in wallets]
    await asyncio.gather(*wallet_loops)


for account in accounts:

    wallets = get_wallets()

    loop = asyncio.get_event_loop()

    if loop.is_closed():
        log.info('Create a new loop')
        loop = asyncio.new_event_loop()

    gp = asyncio.wait([main(account, loop, wallets)])
    loop.run_until_complete(gp)
    loop.close()

The error I get:

2021-07-02 11:45:54.459777 [info     ] Create a new loop

Exception ignored in: <coroutine object throttle.<locals>.run at 0x7f80ce1492b0>
Traceback (most recent call last):
  File "/home/aaa/env/lib/python3.6/site-packages/name/async_support/base/throttle.py", line 51, in run
    future.set_exception(excp)
  File "/usr/lib/python3.6/asyncio/base_events.py", line 591, in call_soon
    self._check_closed()
  File "/usr/lib/python3.6/asyncio/base_events.py", line 377, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

Solution

I guess the canonical thing there would be just to stick your program in an async function and run it all in the event loop:

async def watch_book(account, client, market):
    while True:
        ob = await client.watch_order_book(market.symbol)
        if stop():
            break
        await client.sleep(1000)

async def wallet_loop(account, wallet):
    client = getattr(name, exid)({'enableRateLimit': True, 'asyncio_loop': asyncio.get_running_loop()})
    do_more_stuff()
    ws_loops = [watch_book(account, client, market) for market in markets]
    await asyncio.gather(*ws_loops)
    await client.close()

async def main(account, wallets):
    wallet_loops = [wallet_loop(account, wallet) for wallet in wallets]
    await asyncio.gather(*wallet_loops)
    
async def main_orchestrator(accounts):
    for account in accounts:
        wallets = get_wallets()
        gp = await main(account, wallets)

asyncio.run(main_orchestrator(accounts))

and you could easy even paralelize the processing of all accounts if that makes a difference by changing the function to:

async def main(accounts):
    await asyncio.gather(*(main(account, wallets) for account in accounts)) 

If for some reason you really need each batch to run in a loop,just replace loop = asyncio.get_event_loop() for loop = asyncio.new_event_loop(), nonetheless.

Answered By - jsbueno