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

14 votes480 views6 answers
What is flask
Python Flask
2 votes210 views1 answer
Check if user log in flask
Python
4 votes255 views1 answer
Django slug int url mapping
Python Django
0 vote194 views1 answer
Python dict update
Python
1 vote244 views1 answer
Git clone in python to tmp directory
Python
0 vote196 views1 answer
Django ninja "schema"
Python
23 votes222 views3 answers
Install django rest framework
Python
28 votes224 views6 answers
Convert list to string python
Python
0 vote208 views1 answer
How to find 2 similar words in atext python
Python
0 vote220 views1 answer
Smart Weather Information App Using Python
Python