Python 3.5 asyncio - shutdown all tasks safely using signal handler

asyncio_shutdown_loop.py

import signal
import functools
 
async def looping_task(loop, task_num):
    try:
        while True:
            print('{0}:in looping_task'.format(task_num))
            await asyncio.sleep(5.0, loop=loop)
    except asyncio.CancelledError:
        return "{0}: I was cancelled!".format(task_num)
 
 
async def shutdown(sig, loop):
    print('caught {0}'.format(sig.name))
    tasks = [task for task in asyncio.Task.all_tasks() if task is not
             asyncio.tasks.Task.current_task()]
    list(map(lambda task: task.cancel(), tasks))
    results = await asyncio.gather(*tasks, return_exceptions=True)
    print('finished awaiting cancelled tasks, results: {0}'.format(results))
    loop.stop()
 
 
loop = asyncio.get_event_loop()
for i in range(5):
    asyncio.ensure_future(looping_task(loop, i), loop=loop)
loop.add_signal_handler(signal.SIGTERM,
                        functools.partial(asyncio.ensure_future,
                                          shutdown(signal.SIGTERM, loop)))
try:
    loop.run_forever()
finally:
    loop.close()
 

Thank you, works like a charm.

For catching CTRL+C catch signal.SIGINT

Is there a similar method we might be able to utilize in further python versions? eg: >= 3.8?
This code now raises the following error:

/usr/local/Cellar/python@3.8/3.8.6/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/unix_events.py:140: RuntimeWarning: coroutine 'shutdown' was never awaited
del self._signal_handlers[sig]
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
Traceback (most recent call last):
File "/Users/rjanuary/git/stc-ds-sfbridge/at.py", line 31, in
loop.run_forever()
File "/usr/local/Cellar/python@3.8/3.8.6/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/base_events.py", line 570, in run_forever
self._run_once()
File "/usr/local/Cellar/python@3.8/3.8.6/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/base_events.py", line 1823, in _run_once
event_list = self._selector.select(timeout)
File "/usr/local/Cellar/python@3.8/3.8.6/Frameworks/Python.framework/Versions/3.8/lib/python3.8/selectors.py", line 558, in select
kev_list = self._selector.control(None, max_ev, timeout)
KeyboardInterrupt

Process finished with exit code 130 (interrupted by signal 2: SIGINT)

@rjjanuary Ever figure it out? Exact same thing for me.

Something like this works for me in 3.10. I like nvgoldin's functional style but i find it clunky to read in python and with python-black it's one less line to just do it the imperative way

def add_signal_handlers():
    loop = asyncio.get_event_loop()

    async def shutdown(sig: signal.Signals) -> None:
        """
        Cancel all running async tasks (other than this one) when called.
        By catching asyncio.CancelledError, any running task can perform
        any necessary cleanup when it's cancelled.
        """
        tasks = []
        for task in asyncio.all_tasks(loop):
            if task is not asyncio.current_task(loop):
                task.cancel()
                tasks.append(task)
        results = await asyncio.gather(*tasks, return_exceptions=True)
        print("Finished awaiting cancelled tasks, results: {0}".format(results))
        loop.stop()

    for sig in [signal.SIGINT, signal.SIGTERM]:
        loop.add_signal_handler(sig, lambda: asyncio.create_task(shutdown(sig)))