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

12 votes185 views3 answers
Python check if input is a number
Python
2 votes200 views1 answer
Google calendar Request had insufficient authentication scopes.
Python
1 vote198 views1 answer
Label change in tkinter
Python
1 vote177 views1 answer
Python get latest edited file from any directory
Python
1 vote191 views4 answers
[] python
Python
0 vote204 views1 answer
Custom pylatex command
Python
0 vote192 views4 answers
Comprehension list iloc pandas
Python
0 vote192 views1 answer
Np integral
Python
0 vote221 views1 answer
Item[0]: (i + 1) * 2 for i, item in (sort_loc)
Python
26 votes273 views4 answers
Copy text python
Python