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

2 votes117 views1 answer
ValueError: I/O operation on closed file.
Python
0 vote107 views1 answer
Spread dictionary python
Python
0 vote101 views1 answer
List dictionary to json file python with tab
Python
3 votes139 views6 answers
Django login
Python Django
0 vote143 views1 answer
Python class tostring
Python
0 vote100 views1 answer
Networkx - remove small components from a graph
Python
1 vote131 views1 answer
Python get human readable file size
Python
0 vote110 views1 answer
Python - regexp to find part of an email address
Python
3 votes98 views1 answer
Python procedured
Python
6 votes206 views6 answers
Django Redirect
Python