Handle SIGINT (when sent from another process) and allow binding it to a key binding.

Author: jonathanslenders

prompt_toolkit/application/application.py

@@ -630,6 +630,7 @@ async def run_async(
         self,
         pre_run: Optional[Callable[[], None]] = None,
         set_exception_handler: bool = True,
+        handle_sigint: bool = True,
     ) -> _AppResult:
         """
         Run the prompt_toolkit :class:`~prompt_toolkit.application.Application`
@@ -646,6 +647,9 @@ async def run_async(
         :param set_exception_handler: When set, in case of an exception, go out
             of the alternate screen and hide the application, display the
             exception, and wait for the user to press ENTER.
+        :param handle_sigint: Handle SIGINT signal if possible. This will call
+            the `<sigint>` key binding when a SIGINT is received. (This only
+            works in the main thread.)
         """
         assert not self._is_running, "Application is already running."
 
@@ -781,6 +785,15 @@ async def _run_async2() -> _AppResult:
             self._invalidated = False
 
             loop = get_event_loop()
+
+            if handle_sigint:
+                loop.add_signal_handler(
+                    signal.SIGINT,
+                    lambda *_: loop.call_soon_threadsafe(
+                        self.key_processor.send_sigint
+                    ),
+                )
+
             if set_exception_handler:
                 previous_exc_handler = loop.get_exception_handler()
                 loop.set_exception_handler(self._handle_exception)
@@ -812,12 +825,16 @@ async def _run_async2() -> _AppResult:
                 if set_exception_handler:
                     loop.set_exception_handler(previous_exc_handler)
 
+                    if handle_sigint:
+                        loop.remove_signal_handler(signal.SIGINT)
+
         return await _run_async2()
 
     def run(
         self,
         pre_run: Optional[Callable[[], None]] = None,
         set_exception_handler: bool = True,
+        handle_sigint: bool = True,
         in_thread: bool = False,
     ) -> _AppResult:
         """
@@ -843,6 +860,8 @@ def run(
             `get_appp().create_background_task()`, so that unfinished tasks are
             properly cancelled before the event loop is closed. This is used
             for instance in ptpython.
+        :param handle_sigint: Handle SIGINT signal. Call the key binding for
+            `Keys.SIGINT`. (This only works in the main thread.)
         """
         if in_thread:
             result: _AppResult
@@ -852,7 +871,10 @@ def run_in_thread() -> None:
                 nonlocal result, exception
                 try:
                     result = self.run(
-                        pre_run=pre_run, set_exception_handler=set_exception_handler
+                        pre_run=pre_run,
+                        set_exception_handler=set_exception_handler,
+                        # Signal handling only works in the main thread.
+                        handle_sigint=False,
                     )
                 except BaseException as e:
                     exception = e

prompt_toolkit/application/dummy.py

@@ -24,6 +24,7 @@ def run(
         self,
         pre_run: Optional[Callable[[], None]] = None,
         set_exception_handler: bool = True,
+        handle_sigint: bool = True,
         in_thread: bool = False,
     ) -> None:
         raise NotImplementedError("A DummyApplication is not supposed to run.")
@@ -32,6 +33,7 @@ async def run_async(
         self,
         pre_run: Optional[Callable[[], None]] = None,
         set_exception_handler: bool = True,
+        handle_sigint: bool = True,
     ) -> None:
         raise NotImplementedError("A DummyApplication is not supposed to run.")
 

prompt_toolkit/key_binding/bindings/basic.py

@@ -120,6 +120,7 @@ def load_basic_bindings() -> KeyBindings:
     @handle("insert")
     @handle("s-insert")
     @handle("c-insert")
+    @handle("<sigint>")
     @handle(Keys.Ignore)
     def _ignore(event: E) -> None:
         """

prompt_toolkit/key_binding/key_processor.py

@@ -411,6 +411,13 @@ def flush_keys() -> None:
             self._flush_wait_task.cancel()
         self._flush_wait_task = app.create_background_task(wait())
 
+    def send_sigint(self) -> None:
+        """
+        Send SIGINT. Immediately call the SIGINT key handler.
+        """
+        self.feed(KeyPress(key=Keys.SIGINT), first=True)
+        self.process_keys()
+
 
 class KeyPressEvent:
     """

prompt_toolkit/keys.py

@@ -184,6 +184,8 @@ class Keys(str, Enum):
     WindowsMouseEvent = "<windows-mouse-event>"
     BracketedPaste = "<bracketed-paste>"
 
+    SIGINT = "<sigint>"
+
     # For internal use: key which is ignored.
     # (The key binding for this key should not do anything.)
     Ignore = "<ignore>"

prompt_toolkit/shortcuts/prompt.py

@@ -807,6 +807,7 @@ def _complete_like_readline(event: E) -> None:
             display_completions_like_readline(event)
 
         @handle("c-c", filter=default_focused)
+        @handle("<sigint>")
         def _keyboard_interrupt(event: E) -> None:
             "Abort when Control-C has been pressed."
             event.app.exit(exception=KeyboardInterrupt, style="class:aborting")