Avoid traceback on Ctrl+C during interactive prompt

cylc/flow/scheduler_cli.py

@@ -65,6 +65,7 @@
 )
 from cylc.flow.terminal import (
     cli_function,
+    handle_sigint,
     is_terminal,
     prompt,
 )
@@ -565,15 +566,16 @@ def _version_check(
             ))
             if is_terminal():
                 # we are in interactive mode, ask the user if this is ok
-                options.upgrade = prompt(
-                    cparse(
-                        'Are you sure you want to upgrade from'
-                        f' <yellow>{last_run_version}</yellow>'
-                        f' to <green>{__version__}</green>?'
-                    ),
-                    {'y': True, 'n': False},
-                    process=str.lower,
-                )
+                with handle_sigint():
+                    options.upgrade = prompt(
+                        cparse(
+                            'Are you sure you want to upgrade from'
+                            f' <yellow>{last_run_version}</yellow>'
+                            f' to <green>{__version__}</green>?'
+                        ),
+                        {'y': True, 'n': False},
+                        process=str.lower,
+                    )
                 return options.upgrade
             # we are in non-interactive mode, abort abort abort
             print('Use "--upgrade" to upgrade the workflow.', file=sys.stderr)

cylc/flow/scripts/clean.py

@@ -62,14 +62,25 @@
 import asyncio
 from optparse import SUPPRESS_HELP
 import sys
-from typing import TYPE_CHECKING, Iterable, List, Tuple
+from typing import (
+    TYPE_CHECKING,
+    Iterable,
+    List,
+    Tuple,
+)
 
 from metomi.isodatetime.exceptions import ISO8601SyntaxError
 from metomi.isodatetime.parsers import DurationParser
 
 from cylc.flow import LOG
-from cylc.flow.clean import init_clean, get_contained_workflows
-from cylc.flow.exceptions import CylcError, InputError
+from cylc.flow.clean import (
+    get_contained_workflows,
+    init_clean,
+)
+from cylc.flow.exceptions import (
+    CylcError,
+    InputError,
+)
 import cylc.flow.flags
 from cylc.flow.id_cli import parse_ids_async
 from cylc.flow.loggingutil import set_timestamps
@@ -78,7 +89,12 @@
     CylcOptionParser as COP,
     Options,
 )
-from cylc.flow.terminal import cli_function, is_terminal
+from cylc.flow.terminal import (
+    handle_sigint,
+    cli_function,
+    is_terminal,
+)
+
 
 if TYPE_CHECKING:
     from optparse import Values
@@ -168,12 +184,13 @@ def prompt(workflows: Iterable[str]) -> None:
         print(f'  {workflow}')
 
     if is_terminal():
-        while True:
-            ret = input('Remove these workflows (y/n): ')
-            if ret.lower() == 'y':
-                return
-            if ret.lower() == 'n':
-                sys.exit(1)
+        with handle_sigint():
+            while True:
+                ret = input('Remove these workflows (y/n): ').lower()
+                if ret == 'y':
+                    return
+                if ret == 'n':
+                    sys.exit(1)
     else:
         print(
             "Use --yes to remove multiple workflows in non-interactive mode.",

cylc/flow/scripts/reinstall.py

@@ -98,6 +98,8 @@
 from cylc.flow.terminal import (
     DIM,
     cli_function,
+    handle_sigint,
+    interrupt,
     is_terminal,
 )
 from cylc.flow.workflow_files import (
@@ -226,18 +228,18 @@ async def reinstall_cli(
             display_rose_warning(source)
             display_cylcignore_tip()
             # prompt for permission to continue
-            while usr not in ['y', 'n']:
-                usr = _input(
-                    cparse('<bold>Continue [y/n]: </bold>')
-                ).lower()
+            with handle_sigint(interrupt):
+                while usr not in ['y', 'n']:
+                    usr = _input(
+                        cparse('<bold>Continue [y/n]: </bold>')
+                    ).lower()
 
         else:  # non interactive-mode - no dry-run, no prompt
             usr = 'y'
 
     except KeyboardInterrupt:
         # ensure the "reinstall canceled" message shows for ctrl+c
         usr = 'n'  # cancel the reinstall
-        print()    # clear the traceback line
 
     if usr == 'y':
         # reinstall for real

cylc/flow/terminal.py

@@ -16,20 +16,26 @@
 
 """Functionality to assist working with terminals"""
 
+from contextlib import contextmanager
+from functools import wraps
 import inspect
 import json
 import logging
 import os
+import signal
+from subprocess import (  # nosec
+    PIPE,
+    Popen,
+)
 import sys
-from functools import wraps
-from subprocess import PIPE, Popen  # nosec
 from textwrap import wrap
 from typing import (
     TYPE_CHECKING,
     Any,
     Callable,
     Dict,
     List,
+    NoReturn,
     Optional,
     Sequence,
     TypeVar,
@@ -41,15 +47,18 @@
 from ansimarkup import parse as cparse
 from colorama import init as color_init
 
-import cylc.flow.flags
 from cylc.flow import CYLC_LOG
 from cylc.flow.exceptions import CylcError
+import cylc.flow.flags
 from cylc.flow.loggingutil import CylcLogFormatter
 from cylc.flow.parsec.exceptions import ParsecError
 
 
 if TYPE_CHECKING:
-    from optparse import OptionParser, Values
+    from optparse import (
+        OptionParser,
+        Values,
+    )
 
     T = TypeVar('T')
     StrFunc = Callable[[str], str]
@@ -452,3 +461,29 @@ def flatten_cli_lists(lsts: List[str]) -> List[str]:
         for lst in (lsts or [])
         for item in lst.strip().split(',')
     })
+
+
+@contextmanager
+def handle_sigint(handler: Callable | None = None):
+    """Context manager to handle if Ctrl+C happens while in input().
+
+    If no handler is specified, it will print "Aborted" and exit 1.
+
+    Sets the SIGINT handler inside the context and restores the previous
+    handler after.
+    """
+    prev_handler = signal.signal(signal.SIGINT, handler or abort)
+    try:
+        yield
+    finally:
+        signal.signal(signal.SIGINT, prev_handler)
+
+
+def abort(*args) -> NoReturn:
+    print("\nAborted")
+    sys.exit(1)
+
+
+def interrupt(*args) -> NoReturn:
+    print()  # go to next line after `^C`
+    raise KeyboardInterrupt()