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

39 votes188 views6 answers
Python list contains string
Python
4 votes227 views2 answers
Python regex group
Python
-1 vote198 views2 answers
Encode url
Python
3 votes194 views1 answer
GdScript string format
Python
3 votes188 views2 answers
Enum
Python
0 vote203 views2 answers
Python combine two lists into matrix
Python
3 votes194 views2 answers
Python find largest variable
Python
4 votes286 views5 answers
Python booleans
Python
20 votes195 views4 answers
Python pandas read_excel
Python
2 votes216 views1 answer
Drop a range of rows pandas
Python