Refactored async alerts to use a Queue.

Author: kmvanbrunt

cmd2/cmd2.py

@@ -35,10 +35,12 @@
 import inspect
 import os
 import pydoc
+import queue
 import re
 import sys
 import tempfile
 import threading
+import time
 from code import InteractiveConsole
 from collections import namedtuple
 from collections.abc import (
@@ -48,6 +50,10 @@
     MutableSequence,
     Sequence,
 )
+from dataclasses import (
+    dataclass,
+    field,
+)
 from types import FrameType
 from typing import (
     IO,
@@ -60,7 +66,6 @@
 )
 
 import rich.box
-from prompt_toolkit.application import get_app
 from rich.console import (
     Group,
     RenderableType,
@@ -273,6 +278,19 @@ def remove(self, command_method: CommandFunc) -> None:
             del self._parsers[full_method_name]
 
 
+@dataclass(kw_only=True)
+class AsyncAlert:
+    """Contents of an asynchonous alert which display while user is at prompt.
+
+    :param msg: an optional message to be printed above the prompt.
+    :param prompt: an optional string to dynamically replace the current prompt.
+    """
+
+    msg: str | None = None
+    prompt: str | None = None
+    timestamp: float = field(default_factory=time.monotonic, init=False)
+
+
 class Cmd:
     """An easy but powerful framework for writing line-oriented command interpreters.
 
@@ -587,6 +605,13 @@ def __init__(
         # Command parsers for this Cmd instance.
         self._command_parsers: _CommandParsers = _CommandParsers(self)
 
+        # Members related to printing asychronous alerts
+        self.alert_queue: queue.Queue[AsyncAlert] = queue.Queue()
+        self._alerter_gate = threading.Event()
+        self._alerter_shutdown = threading.Event()
+        self._process_alerts_thread: threading.Thread | None = None
+        self._prompt_drawn_at: float = 0.0  # Uses time.monotonic()
+
         # Add functions decorated to be subcommands
         self._register_subcommands(self)
 
@@ -2588,7 +2613,7 @@ def pre_prompt(self) -> None:
         """Ran just before the prompt is displayed (and after the event loop has started)."""
 
     def precmd(self, statement: Statement | str) -> Statement:
-        """Ran just before the command is executed by [cmd2.Cmd.onecmd][] and after adding it to history (cmd  Hook method).
+        """Ran just before the command is executed by [cmd2.Cmd.onecmd][] and after adding it to history (cmd Hook method).
 
         :param statement: subclass of str which also contains the parsed input
         :return: a potentially modified version of the input Statement object
@@ -3200,9 +3225,9 @@ def _read_raw_input(
     ) -> str:
         """Execute the low-level input read from either a terminal or a redirected stream.
 
-        If the session is interactive (TTY), it uses `prompt_toolkit` to render a
-        rich UI with completion and `patch_stdout` protection. If non-interactive
-        (Pipe/File), it performs a direct line read from `stdin`.
+        If input is coming from a TTY, it uses `prompt_toolkit` to render a
+        UI with completion and `patch_stdout` protection. Otherwise it performs
+        a direct line read from `stdin`.
 
         :param prompt: the prompt text or a callable that returns the prompt.
         :param session: the PromptSession instance to use for reading.
@@ -3321,6 +3346,42 @@ def read_input(
 
         return self._read_raw_input(prompt, temp_session, completer_to_use)
 
+    def _process_alerts(self) -> None:
+        """Background worker that processes queued alerts and prompt updates.
+
+        This loop waits for the prompt gate to open, ensuring that background
+        messages and UI refreshes only occur while the user is at an
+        interactive prompt, avoiding interference with active commands.
+        """
+        while not self._alerter_shutdown.is_set():
+            try:
+                # Wait for an alert
+                alert = self.alert_queue.get(timeout=0.5)
+
+                # Block if not at a prompt
+                while not self._alerter_gate.is_set():
+                    if self._alerter_shutdown.is_set():
+                        return
+                    self._alerter_gate.wait(timeout=0.1)
+
+                # Print and update
+                with patch_stdout():
+                    if alert.msg:
+                        print(alert.msg)
+
+                # Only update if the alert was generated after the current prompt was drawn on the screen.
+                if (alert.prompt is not None and
+                    alert.prompt != self.prompt and
+                    alert.timestamp > self._prompt_drawn_at):  # fmt: skip
+                    self.prompt = alert.prompt
+
+                    # Don't update the UI if we are at a continuation prompt.
+                    if not self._at_continuation_prompt:
+                        self.session.app.invalidate()
+
+            except queue.Empty:  # noqa: PERF203
+                continue
+
     def _read_command_line(self, prompt: str) -> str:
         """Read the next command line from the input stream.
 
@@ -3331,19 +3392,39 @@ def _read_command_line(self, prompt: str) -> str:
         """
 
         # Use dynamic prompt if the prompt matches self.prompt
-        def get_prompt() -> ANSI | str:
+        def get_prompt() -> ANSI:
             return ANSI(self.prompt)
 
         prompt_to_use: Callable[[], ANSI | str] | ANSI | str = ANSI(prompt)
         if prompt == self.prompt:
             prompt_to_use = get_prompt
 
-        return self._read_raw_input(
-            prompt=prompt_to_use,
-            session=self.session,
-            completer=self.completer,
-            pre_run=self.pre_prompt,
-        )
+        def _pre_prompt() -> None:
+            """Run standard pre-prompt processing and activate the background alerter."""
+            self.pre_prompt()
+
+            # Record exactly when the user was presented with this prompt
+            self._prompt_drawn_at = time.monotonic()
+
+            # Start alerter thread if it's not already running
+            if self._process_alerts_thread is None or not self._process_alerts_thread.is_alive():
+                self._alerter_shutdown.clear()
+                self._process_alerts_thread = threading.Thread(target=self._process_alerts, daemon=True)
+                self._process_alerts_thread.start()
+
+            # Allow alerts to be printed
+            self._alerter_gate.set()
+
+        try:
+            return self._read_raw_input(
+                prompt=prompt_to_use,
+                session=self.session,
+                completer=self.completer,
+                pre_run=_pre_prompt,
+            )
+        finally:
+            # Ensure no alerts print while the command is processing
+            self._alerter_gate.clear()
 
     def _cmdloop(self) -> None:
         """Repeatedly issue a prompt, accept input, parse it, and dispatch to apporpriate commands.
@@ -5207,66 +5288,19 @@ def do__relative_run_script(self, args: argparse.Namespace) -> bool | None:
         # self.last_result will be set by do_run_script()
         return self.do_run_script(su.quote(relative_path))
 
-    def async_alert(self, alert_msg: str, new_prompt: str | None = None) -> None:
-        """Display an important message to the user while they are at a command line prompt.
-
-        To the user it appears as if an alert message is printed above the prompt and their
-        current input text and cursor location is left alone.
-
-        This function checks self._in_prompt to ensure a prompt is on screen.
-        If the main thread is not at the prompt, a RuntimeError is raised.
+    def add_alert(self, *, msg: str | None = None, prompt: str | None = None) -> None:
+        """Thread-safe method to request UI updates.
 
-        This function is only needed when you need to print an alert or update the prompt while the
-        main thread is blocking at the prompt. Therefore, this should never be called from the main
-        thread. Doing so will raise a RuntimeError.
+        :param msg: an optional message to be printed above the prompt.
+        :param prompt: an optional string to dynamically replace the current prompt.
 
-        :param alert_msg: the message to display to the user
-        :param new_prompt: If you also want to change the prompt that is displayed, then include it here.
-                           See async_update_prompt() docstring for guidance on updating a prompt.
-        :raises RuntimeError: if called from the main thread.
-        :raises RuntimeError: if main thread is not currently at the prompt.
-        """
-
-        # Check if prompt is currently displayed and waiting for user input
-        def _alert() -> None:
-            if new_prompt is not None:
-                self.prompt = new_prompt
-
-            if alert_msg:
-                # Since we are running in the loop, patch_stdout context manager from read_input
-                # should be active (if tty), or at least we are in the main thread.
-                print(alert_msg)
-
-            if hasattr(self, 'session'):
-                # Invalidate to force prompt update
-                get_app().invalidate()
-
-        # Schedule the alert to run on the main thread's event loop
-        try:
-            get_app().loop.call_soon_threadsafe(_alert)  # type: ignore[union-attr]
-        except AttributeError:
-            # Fallback if loop is not accessible (e.g. prompt not running or session not initialized)
-            # This shouldn't happen if _in_prompt is True, unless prompt exited concurrently.
-            raise RuntimeError("Event loop not available") from None
-
-    def async_update_prompt(self, new_prompt: str) -> None:  # pragma: no cover
-        """Update the command line prompt while the user is still typing at it.
-
-        This is good for alerting the user to system changes dynamically in between commands.
-        For instance you could alter the color of the prompt to indicate a system status or increase a
-        counter to report an event. If you do alter the actual text of the prompt, it is best to keep
-        the prompt the same width as what's on screen. Otherwise the user's input text will be shifted
-        and the update will not be seamless.
-
-        If user is at a continuation prompt while entering a multiline command, the onscreen prompt will
-        not change. However, self.prompt will still be updated and display immediately after the multiline
-        line command completes.
-
-        :param new_prompt: what to change the prompt to
-        :raises RuntimeError: if called from the main thread.
-        :raises RuntimeError: if main thread is not currently at the prompt.
+        1. print an alert: add_alert(msg="System error!")
+        2. print and update prompt: add_alert(msg="Logged in", prompt="user@host> ")
+        3. update prompt only: add_alert(prompt="waiting> ")
         """
-        self.async_alert('', new_prompt)
+        if msg is not None or prompt is not None:
+            alert = AsyncAlert(msg=msg, prompt=prompt)
+            self.alert_queue.put(alert)
 
     @staticmethod
     def set_window_title(title: str) -> None:  # pragma: no cover