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 votes146 views1 answer
Python for-while
Python
0 vote169 views1 answer
Get data from model with field name in django
Python Django
0 vote199 views1 answer
Shebang line python
Python
1 vote175 views2 answers
How to remove the very last character of a text file in python
Python
0 vote174 views1 answer
Django password field encryption
Python Django
3 votes178 views2 answers
Split imagedatagenerator into x_train and y_train
Python
33 votes171 views5 answers
How to filter out all NaN values in pandas df
Python
0 vote172 views1 answer
Regex remove all html tags except br python
Python
1 vote117 views1 answer
For i in range(0,4):for i in range(0,4):for i in range(0,4):for i in range(0,4):for i in range(0,4):for i in range(0,4):for i in range(0,4):for i in range(0,4):for i in range(0,4):
Python
1 vote141 views1 answer
Python sqlite dict
Python