Patch loop runner _run_once

import asyncio
def _run_once(self):
	num_tasks = len(self._scheduled)
    print("num tasks in queue: {}".format(num_tasks))
    super(asyncio.SelectorEventLoop, self)._run_once()

EventLoop = asyncio.SelectorEventLoop
EventLoop._run_once = _run_once
loop = EventLoop()
asyncio.set_event_loop(loop)
async def task(n):
    await asyncio.sleep(n)
    print("sleep: {} sec".format(n))

coro = loop.create_task(task(3))
loop.run_until_complete(coro)
# num tasks in queue: 0
# num tasks in queue: 1
# num tasks in queue: 0
# sleep: 3 sec
# num tasks in queue: 0
loop.close()

Source: pythonsheets.com

Similar codes

0 vote191 views1 answer
Connectionrefusederror at /accounts/signup/ django allauth
Python
25 votes194 views6 answers
Python remove spaces
Python
0 vote193 views1 answer
Gitpod how to execute python file
Python
0 vote187 views1 answer
Jupyter notebook not showing all columns
Python
5 votes195 views6 answers
Plt.xticks
Python
1 vote201 views2 answers
Can you look for specific characters in python
Python
0 vote188 views1 answer
Csr_matric scipy lib
Python
0 vote181 views1 answer
Code runner runs python 2
Python
1 vote196 views1 answer
Matplotlib dark mode
Python
2 votes188 views2 answers
Horizontal bar plot python
Python