Issue
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()
Solution
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 By - Sy Ker
0 comments:
Post a Comment
Note: Only a member of this blog may post a comment.