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 vote252 views1 answer
Get fields in object python
Python
1 vote249 views1 answer
Python nameerror input
Python
0 vote293 views1 answer
Fastest clicker python
Python
1 vote276 views1 answer
Selenium scroll element into view inside overflow python
Python
0 vote263 views1 answer
Get second min no from array in python
Python
18 votes285 views5 answers
Pandas replace na with 0
Python
0 vote320 views1 answer
At=error code=h10 desc= app crashed python flask
Python Flask
0 vote285 views1 answer
Plotly ylog
Python
0 vote265 views1 answer
Abrir notebooks jupyter administrador de archivos
Python
0 vote271 views1 answer
2 distinct numbers random number generator python
Python