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

4 votes275 views1 answer
Remove specific word from string using python
Python
0 vote287 views1 answer
Python close database connection
Python
6 votes316 views1 answer
How to set the icon of the window in pygame
Python
0 vote291 views1 answer
Destroy label tkinter
Python
11 votes307 views2 answers
Add pip to path
Python
0 vote290 views1 answer
Python type hint class itself
Python
0 vote281 views1 answer
Feature scaling in python
Python
2 votes271 views1 answer
Edge detection opencv python
Python
0 vote299 views1 answer
How to change dtype object to int
Python
0 vote317 views1 answer
Van first name van second name van last name
Python