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

0 vote371 views2 answers
From django.core.management import execute_from_command_line ImportError: No module named django.core.management
Python
12 votes394 views5 answers
Python async await
Python
0 vote389 views1 answer
ImportError: cannot import name 'config' from 'decouple' (C:\Users\Yemi\AppData\Local\Programs\Python\Python310\lib\site-packages\decouple\__init__.py)
Python
0 vote337 views1 answer
Hex python add 0
Python
0 vote327 views1 answer
Create database tables python
Python
2 votes367 views2 answers
Discord.py ping command
Python
4 votes325 views3 answers
Python vs Java
Python
0 vote358 views2 answers
Python web scraping project
Python
17 votes344 views6 answers
How to append to a list in python
Python
1 vote319 views1 answer
Combination in python math
Python