Python async and exception handling

Question

Below is a small example of something that I need to do in a much more involved, real-world form. I need my program to close in an orderly fashion, shutting down connections cleanly, if a keyboard interrupt (^C) is pressed. But as the results below show, the keyboard interrupt isn't being caught. The "finally" works but the "except" doesn't. And if I catch it at the main level instead of where my try statement is, I will no longer have the context of my function to display the information I might like to display.

Is there a better way to do this?

async def tryAsynchronous():
    count = 0
    try:
        while True:
            count = count + 1
            print(count)
            await asyncio.sleep(1)
    except KeyboardInterrupt as e:
        print("interrupt", count)
    finally:
        print("final count", count)

asyncio.run(tryAsynchronous())

Here are the results, when I press control-C after the third count:

% python3 kbint.py
1
2
3
^Cfinal count 3
Traceback (most recent call last):
  File "/Users/ken/thermo/py/test/kbint.py", line 45, in <module>
    asyncio.run(tryAsynchronous())
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/base_events.py", line 629, in run_until_complete
    self.run_forever()
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/base_events.py", line 596, in run_forever
    self._run_once()
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/base_events.py", line 1854, in _run_once
    event_list = self._selector.select(timeout)
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/selectors.py", line 562, in select
    kev_list = self._selector.control(None, max_ev, timeout)
KeyboardInterrupt

does the answer and comment here help? https://stackoverflow.com/a/54528397/202168 — Anentropic, Jul 07 '22 at 14:44

score 1 · Answer 1 · answered Jul 07 '22 at 14:46

How about using signal to handle the quit, in the following codes, a global flg_run is used to tell the coroutine to keep running or not:

import asyncio
import signal

flg_run = True


def sigint_handler(signal, frame):
    global flg_run
    print('KeyboardInterrupt is caught')
    flg_run = False


signal.signal(signal.SIGINT, sigint_handler)


async def tryAsynchronous():
    global flg_run
    count = 0
    while flg_run:
        count = count + 1
        print(count)
        await asyncio.sleep(1)
    print("interrupt", count)
    print("final count", count)


asyncio.run(tryAsynchronous())

The output looks like:

1
2
3
4
KeyboardInterrupt is caught
interrupt 4
final count 4

Python async and exception handling

1 Answers1