gracefully shutdown the top asyncio coroutine in Python

example.py

import asyncio
import signal
import logging
import contextlib

logger = logging.getLogger(__name__)


def asyncio_run(coro):
    task = asyncio.ensure_future(coro)
    loop = asyncio.get_event_loop()

    def canceller_maker(reason):
        def canceller():
            logger.info(reason)
            task.cancel()

        return canceller

    loop.add_signal_handler(
        sig=signal.SIGTERM, callback=canceller_maker("cancelled by user via SIGTERM")
    )
    loop.add_signal_handler(
        sig=signal.SIGINT, callback=canceller_maker("cancelled by user via SIGINT")
    )
    try:
        with contextlib.suppress(asyncio.CancelledError):
            return loop.run_until_complete(task)
    finally:
        loop.close()


 async def top():
    try:
        await asyncio.sleep(10000)
    except asyncio.CancelledError:
        print("cancelled")
        raise
        
asyncio_run(top())