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 vote269 views1 answer
Consider a string note: "welcome" statment will rais error
Python
3 votes325 views1 answer
Csrf token exempt django
Python Django
0 vote304 views2 answers
Python:Gann square of 9
Python
1 vote288 views1 answer
Pandas dataframe lists as columns
Python
0 vote341 views1 answer
Mechanize python #3
Python
0 vote292 views1 answer
Django form is onvalid
Python Django
-1 vote303 views1 answer
Splash screen in python tkinter
Python
0 vote296 views1 answer
My_list = ['a','b'] my_list.append('c') print(my_list)
Python
2 votes308 views2 answers
Say command python
Python
0 vote278 views1 answer
Is python platform independent
Python