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()
 

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)))