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

8 votes330 views2 answers
Find max in a dataframe
Python
0 vote387 views1 answer
Free download django app for windows 10
Python Django
8 votes334 views2 answers
Python iterate list reverse
Python
0 vote393 views1 answer
Django model plural
Python Django
19 votes350 views4 answers
Install python on ubuntu
Python
0 vote361 views1 answer
Standard destructuring assignments in python
Python
21 votes353 views4 answers
Convert radians to degrees python
Python
0 vote392 views1 answer
Snake game using python
Python
17 votes315 views1 answer
Strip all spaces python
Python
0 vote306 views1 answer
Glob.iglob sort path
Python