时间戳

首页 > 解决方案 > 如何从异步信号处理程序中捕获自定义异常？

python - 如何从异步信号处理程序中捕获自定义异常？

问题描述

在使用asyncio.

ShutdownApp如果我从下面扔do_io()，我可以正确地抓住它run_app()。但是，当从 引发异常时handle_sig()，我似乎无法捕捉到它。

使用 Python 3.8.5 测试的最小的、可重现的示例：

import asyncio
from functools import partial
import os
import signal
from signal import Signals


class ShutdownApp(BaseException):
    pass


os.environ["PYTHONASYNCIODEBUG"] = "1"


class App:
    def __init__(self):
        self.loop = asyncio.get_event_loop()

    def _add_signal_handler(self, signal, handler):
        self.loop.add_signal_handler(signal, handler, signal)

    def setup_signals(self) -> None:
        self._add_signal_handler(signal.SIGINT, self.handle_sig)

    def handle_sig(self, signum):
        print(f"\npid: {os.getpid()}, Received signal: {Signals(signum).name}, raising error for exit")
        raise ShutdownApp("Exiting")

    async def do_io(self):
        print("io start. Press Ctrl+C now.")
        await asyncio.sleep(5)
        print("io end")

    def run_app(self):
        print("Starting Program")
        try:
            self.loop.run_until_complete(self.do_io())
        except ShutdownApp as e:
            print("ShutdownApp caught:", e)
            # TODO: do other shutdown related items
        except:
            print("Other error")
        finally:
            self.loop.close()


if __name__ == "__main__":
    my_app = App()
    my_app.setup_signals()
    my_app.run_app()
    print("Finished")

使用 asyncio 调试模式按下CTRL+C(for ) 后的输出：SIGINT

(env_aiohttp) anav@anav-pc:~/Downloads/test$ python test_asyncio_signal.py 
Starting Program
io start. Press Ctrl+C now.
^C
pid: 20359, Received signal: SIGINT, raising error for exit
Exception in callback App.handle_sig(<Signals.SIGINT: 2>)
handle: <Handle App.handle_sig(<Signals.SIGINT: 2>) created at /home/anav/miniconda3/envs/env_aiohttp/lib/python3.8/asyncio/unix_events.py:99>
source_traceback: Object created at (most recent call last):
  File "test_asyncio_signal.py", line 50, in <module>
    my_app.setup_signals()
  File "test_asyncio_signal.py", line 25, in setup_signals
    self._add_signal_handler(signal.SIGINT, self.handle_sig)
  File "test_asyncio_signal.py", line 22, in _add_signal_handler
    self.loop.add_signal_handler(signal, handler, signal)
  File "/home/anav/miniconda3/envs/env_aiohttp/lib/python3.8/asyncio/unix_events.py", line 99, in add_signal_handler
    handle = events.Handle(callback, args, self, None)
Traceback (most recent call last):
  File "/home/anav/miniconda3/envs/env_aiohttp/lib/python3.8/asyncio/events.py", line 81, in _run
    self._context.run(self._callback, *self._args)
  File "test_asyncio_signal.py", line 31, in handle_sig
    raise ShutdownApp("Exiting")
ShutdownApp: Exiting
io end
Finished

预期输出：

Starting Program
io start. Press Ctrl+C now.
^C
pid: 20359, Received signal: SIGINT, raising error for exit
ShutdownApp caught: Exiting
io end
Finished

是否可以从信号处理程序中引发自定义异常asyncio？如果是这样，我如何正确捕获/排除它？

标签： pythonpython-3.xpython-asynciouvloop

解决方案

handle_sig是一个回调，所以它直接在事件循环之外运行，并且它的异常只是通过一个全局钩子报告给用户。如果您希望那里引发的异常在程序的其他地方被捕获，您需要使用未来将异常从handle_sig您希望它注意到的地方转移。

要在顶层捕获异常，您可能需要引入另一种方法，我们称之为它async_main()，它等待或 self.do_io()之前创建的未来完成：

    def __init__(self):
        self.loop = asyncio.get_event_loop()
        self.done_future = self.loop.create_future()

    async def async_main(self):
        # wait for do_io or done_future, whatever happens first
        io_task = asyncio.create_task(self.do_io())
        await asyncio.wait([self.done_future, io_task],
                           return_when=asyncio.FIRST_COMPLETED)
        if self.done_future.done():
            io_task.cancel()
            await self.done_future  # propagate the exception, if raised
        else:
            self.done_future.cancel()

要从内部引发异常handle_sig，您只需在未来对象上设置异常：

    def handle_sig(self, signum):
        print(f"\npid: {os.getpid()}, Received signal: {Signals(signum).name}, raising error for exit")
        self.done_future.set_exception(ShutdownApp("Exiting"))

最后，修改run_app为 pass self.async_main()to run_until_complete，一切就绪：

$ python3 x.py
Starting Program
io start. Press Ctrl+C now.
^C
pid: 2069230, Received signal: SIGINT, raising error for exit
ShutdownApp caught: Exiting
Finished

最后，请注意可靠地捕获键盘中断是一项非常棘手的任务，并且上面的代码可能无法涵盖所有​​极端情况。

推荐阅读