python3.8 RuntimeError: no running event loop

Question

I have taken the below code snippet from a book by author caleb hattingh. I tried running the code snippet and faced this error.(Practicing)

How do I resolve this?

import asyncio

async def f(delay):
    await asyncio.sleep(1 / delay)
    return delay

loop = asyncio.get_event_loop()
for i in range(10):
    loop.create_task(f(i))

print(loop)
pending = asyncio.all_tasks()
group = asyncio.gather(*pending, return_exceptions=True)
results = loop.run_until_complete(group)
print(f'Results: {results}')
loop.close()

Does this answer your question? Asyncio in corroutine RuntimeError: no running event loop — Tamil Selvan
– Tamil Selvan, Commented Jan 27, 2022 at 8:31

Sy Ker · Accepted Answer · 2022-01-27 08:35:13Z

6

You have to pass the loop as argument to the .all_tasks() function:

pending = asyncio.all_tasks(loop)

Output:

<_UnixSelectorEventLoop running=False closed=False debug=False>
<_GatheringFuture pending>
Results: [8, 5, 2, 9, 6, 3, ZeroDivisionError('division by zero'), 7, 4, 1]

So for a full correction of your script:

import asyncio

async def f(delay):
    if delay:
        await asyncio.sleep(1 / delay)
    return delay

loop = asyncio.get_event_loop()
for i in range(10):
    loop.create_task(f(i))

print(loop)
pending = asyncio.all_tasks(loop)
group = asyncio.gather(*pending, return_exceptions=True)
results = loop.run_until_complete(group)
print(f'Results: {results}')
loop.close()

answered Jan 27, 2022 at 8:35

Sy Ker

2,2081 gold badge7 silver badges22 bronze badges

Sign up to request clarification or add additional context in comments.

Comments

Michael Szczesny · Accepted Answer · 2022-01-27 09:24:56Z

1

Using python3.7 and later versions you can omit the explicit creation and management of the event loop and tasks. The asyncio API has changed a few times and you will find tutorials covering outdated syntax. The following implementation corresponds to your solution.

import asyncio

async def f(delay):
    await asyncio.sleep(1 / delay)
    return delay

async def main():
    return await asyncio.gather(*[f(i) for i in range(10)], return_exceptions=True)

print(asyncio.run(main()))

Output

[ZeroDivisionError('division by zero'), 1, 2, 3, 4, 5, 6, 7, 8, 9]

answered Jan 27, 2022 at 9:24

community wiki

Michael Szczesny

1 Comment

Shu Over a year ago

HI, very usable solution. BTW, if we want to manually craete task using asyncio.create_task(), this error still apears, is there any workaround for this?

Collectives™ on Stack Overflow

python3.8 RuntimeError: no running event loop

2 Answers 2

Comments

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related