Close asyncio loop on KeyboardInterrupt - Run stop routine

You want to add a signal handler as shown in this example in the docs:

import asyncio
import functools
import os
import signal

def ask_exit(signame, loop):
    print("got signal %s: exit" % signame)
    loop.stop()

async def main():
    loop = asyncio.get_running_loop()

    for signame in {'SIGINT', 'SIGTERM'}:
        loop.add_signal_handler(
            getattr(signal, signame),
            functools.partial(ask_exit, signame, loop))

    await asyncio.sleep(3600)

print("Event loop running for 1 hour, press Ctrl+C to interrupt.")
print(f"pid {os.getpid()}: send SIGINT or SIGTERM to exit.")

asyncio.run(main())

That's a bit of an overcomplicated/outdated example though, consider it more like this (your coroutine code goes where the asyncio.sleep call is):

import asyncio
from signal import SIGINT, SIGTERM
    

async def main():
    loop = asyncio.get_running_loop()
    for signal_enum in [SIGINT, SIGTERM]:
        loop.add_signal_handler(signal_enum, loop.stop)

    await asyncio.sleep(3600) # Your code here


asyncio.run(main())

At this point a Ctrl + C will break the loop and raise a RuntimeError, which you can catch by putting the asyncio.run call in a try/except block like so:

try:
    asyncio.run(main())
except RuntimeError as exc:
    expected_msg = "Event loop stopped before Future completed."
    if exc.args and exc.args[0] == expected_msg:
        print("Bye")
    else:
        raise

That's not very satisfying though (what if something else caused the same error?), so I'd prefer to raise a distinct error. Also, if you're exiting on the command line, the proper thing to do is to return the proper exit code (in fact, the code in the example just uses the name, but it's actually an IntEnum with that numeric exit code in it!)

import asyncio
from functools import partial
from signal import SIGINT, SIGTERM
from sys import stderr

class SignalHaltError(SystemExit):
    def __init__(self, signal_enum):
        self.signal_enum = signal_enum
        print(repr(self), file=stderr)
        super().__init__(self.exit_code)

    @property
    def exit_code(self):
        return self.signal_enum.value

    def __repr__(self):
        return f"\nExitted due to {self.signal_enum.name}"

def immediate_exit(signal_enum, loop):
    loop.stop()
    raise SignalHaltError(signal_enum=signal_enum)

async def main():
    loop = asyncio.get_running_loop()

    for signal_enum in [SIGINT, SIGTERM]:
        exit_func = partial(immediate_exit, signal_enum=signal_enum, loop=loop)
        loop.add_signal_handler(signal_enum, exit_func)

    await asyncio.sleep(3600)

print("Event loop running for 1 hour, press Ctrl+C to interrupt.")

asyncio.run(main())

Which when Ctrl + C'd out of gives:

python cancelling_original.py

⇣

Event loop running for 1 hour, press Ctrl+C to interrupt.
^C
Exitted due to SIGINT

echo $?

⇣

2

Now there's some code I'd be happy to serve! :^)

P.S. here it is with type annotations:

from __future__ import annotations

import asyncio
from asyncio.events import AbstractEventLoop
from functools import partial
from signal import Signals, SIGINT, SIGTERM
from sys import stderr
from typing import Coroutine

class SignalHaltError(SystemExit):
    def __init__(self, signal_enum: Signals):
        self.signal_enum = signal_enum
        print(repr(self), file=stderr)
        super().__init__(self.exit_code)

    @property
    def exit_code(self) -> int:
        return self.signal_enum.value

    def __repr__(self) -> str:
        return f"\nExitted due to {self.signal_enum.name}"

def immediate_exit(signal_enum: Signals, loop: AbstractEventLoop) -> None:
    loop.stop()
    raise SignalHaltError(signal_enum=signal_enum)

async def main() -> Coroutine:
    loop = asyncio.get_running_loop()

    for signal_enum in [SIGINT, SIGTERM]:
        exit_func = partial(immediate_exit, signal_enum=signal_enum, loop=loop)
        loop.add_signal_handler(signal_enum, exit_func)

    return await asyncio.sleep(3600)

print("Event loop running for 1 hour, press Ctrl+C to interrupt.")

asyncio.run(main())

The advantage of a custom exception here is that you can then catch it specifically, and avoid the traceback being dumped to the screen

try:
    asyncio.run(main())
except SignalHaltError as exc:
    # log.debug(exc)
    pass
else:
    raise