Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

- Bug Fixes
- Fixed a couple type issues that emerged with the upgrade to `mypy` 2.2.0
- Enhancements
- Async alert messages can now be Rich renderables.

## 4.1.0 (July 7, 2026)

Expand Down
43 changes: 35 additions & 8 deletions cmd2/cmd2.py
Original file line number Diff line number Diff line change
Expand Up @@ -312,15 +312,20 @@ def remove(self, command_method: BoundCommandFunc) -> None:
class AsyncAlert:
"""Contents of an asynchronous alert which display while user is at prompt.

:param msg: an optional message to be printed above the prompt.
:param msg: an optional printable object (including Rich renderables) to be
printed above the prompt.
:param soft_wrap: Enable soft wrap mode. This only applies with msg is not None.
Defaults to True. See print_to() docstring for more details on
this parameter.
:param prompt: an optional string to dynamically replace the current prompt.

:ivar timestamp: monotonic creation time of the alert. If an alert was created
before the current prompt was rendered, its prompt data is ignored
to avoid a stale display, but its msg data will still be displayed.
"""

msg: str | None = None
msg: Any | None = None
soft_wrap: bool = True
prompt: str | None = None
timestamp: float = field(default_factory=time.monotonic, init=False)

Expand Down Expand Up @@ -1572,7 +1577,7 @@ def print_to(
If False, Rich wraps text to fit the terminal width.
Set this to False when printing structured Renderables like
Tables, Panels, or Columns to ensure they render as expected.
For example, when soft_wrap is True Panels truncate text
For example, when soft_wrap is True, Panels truncate text
which is wider than the terminal.
:param justify: justify method ("left", "center", "right", "full"). Defaults to None.
:param emoji: If True, Rich will replace emoji codes (e.g., :smiley:) with their
Expand Down Expand Up @@ -3657,7 +3662,19 @@ def _process_alerts(self) -> None:
# prevents async alerts from printing once a command starts.

# Print all alerts at once to reduce flicker.
alert_text = "\n".join(alert.msg for alert in self._alert_queue if alert.msg)
console = self._get_core_print_console(
file=self.stdout,
emoji=False,
markup=False,
highlight=False,
)

with console.capture() as capture:
for alert in self._alert_queue:
if alert.msg is not None:
console.print(alert.msg, soft_wrap=alert.soft_wrap)

alert_text = capture.get()

# Find the latest prompt update among all pending alerts.
latest_prompt = None
Expand All @@ -3681,7 +3698,7 @@ def _process_alerts(self) -> None:
if alert_text:
# Print the alert messages above the prompt.
with patch_stdout():
print_formatted_text(pt_filter_style(alert_text))
print_formatted_text(ANSI(alert_text), end="")

elif latest_prompt is not None:
# Refresh UI immediately to show the new prompt
Expand Down Expand Up @@ -5616,23 +5633,33 @@ 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 add_alert(self, *, msg: str | None = None, prompt: str | None = None) -> None:
def add_alert(
self,
*,
msg: Any | None = None,
soft_wrap: bool = True,
prompt: str | None = None,
) -> None:
"""Queue an asynchronous alert to be displayed when the prompt is active.

Examples:
add_alert(msg="System error!") # Print message only
add_alert(prompt="user@host> ") # Update prompt only
add_alert(msg="Done", prompt="> ") # Update both

:param msg: an optional message to be printed above the prompt.
:param msg: an optional printable object (including Rich renderables) to be
printed above the prompt.
:param soft_wrap: Enable soft wrap mode. This only applies with msg is not None.
Defaults to True. See print_to() docstring for more details on
this parameter.
:param prompt: an optional string to dynamically replace the current prompt.

"""
if msg is None and prompt is None:
return

with self._alert_condition:
alert = AsyncAlert(msg=msg, prompt=prompt)
alert = AsyncAlert(msg=msg, soft_wrap=soft_wrap, prompt=prompt)
self._alert_queue.append(alert)
self._alert_condition.notify_all()

Expand Down
161 changes: 80 additions & 81 deletions examples/async_printing.py
Original file line number Diff line number Diff line change
@@ -1,37 +1,67 @@
#!/usr/bin/env python
"""A simple example demonstrating an application that asynchronously prints alerts, updates the prompt
and changes the window title.
"""A simple example demonstrating an application that asynchronously prints alerts,
updates the prompt, and changes the window title.
"""

import secrets
import threading
import time
from typing import Any

from rich.console import Group
from rich.panel import Panel
from rich.style import Style
from rich.table import Table
from rich.text import Text

import cmd2
from cmd2 import (
Color,
stylize,
)

ALERTS = [
"Watch as this application prints alerts and updates the prompt",
"This will only happen when the prompt is present",
"Notice how it doesn't interfere with your typing or cursor location",
"Go ahead and type some stuff and move the cursor throughout the line",
"Keep typing...",
"Move that cursor...",
"Pretty seamless, eh?",
"Feedback can also be given in the window title. Notice the alert count up there?",
"You can stop and start the alerts by typing stop_alerts and start_alerts",
"This demo will now continue to print alerts at random intervals",
]

def get_alerts() -> list[tuple[Any, bool]]:
"""Return a list of (alert_msg, soft_wrap) tuples."""
table = Table("Command", "Description", title="Quick Help")
table.add_row("start_alerts", "Start the async alert generator")
table.add_row("stop_alerts", "Stop the async alert generator")
table.add_row("help", "Show help menu")

# Set soft_wrap to False when printing structured Renderables like Tables, Panels, or Columns
# to ensure they render as expected. For example, when soft_wrap is True, Panels truncate
# text which is wider than the terminal.
return [
(
Text(
"Watch as this application prints alerts asynchronously!",
style=Style(
color=Color.BRIGHT_CYAN,
bold=True,
),
),
True,
),
("Notice how alerts don't interfere with your typing or cursor location.", True),
(
Panel(
"This message is wrapped in a Rich Panel!",
title="System Alert",
border_style=Color.BRIGHT_BLUE,
expand=False,
),
False,
),
(table, False),
("You can stop and start the alerts by typing stop_alerts and start_alerts.", True),
]


class AlerterApp(cmd2.Cmd):
"""An app that shows off async_alert() and async_update_prompt()."""
"""An app that shows off add_alert() and set_window_title()."""

def __init__(self) -> None:
"""Initializer."""
"""Initialize AlerterApp."""
super().__init__()

self.prompt = "(APR)> "
Expand All @@ -40,7 +70,6 @@ def __init__(self) -> None:
self._stop_event = threading.Event()
self._add_alert_thread = threading.Thread()
self._alert_count = 0
self._next_alert_time = 0.0

# Create some hooks to handle the starting and stopping of our thread
self.register_preloop_hook(self._preloop_hook)
Expand All @@ -56,13 +85,13 @@ def _preloop_hook(self) -> None:
self._add_alert_thread.start()

def _postloop_hook(self) -> None:
"""Stops the alerter thread."""
"""Stop the alerter thread."""
self._stop_event.set()
if self._add_alert_thread.is_alive():
self._add_alert_thread.join()

def do_start_alerts(self, _: cmd2.Statement) -> None:
"""Starts the alerter thread."""
"""Start the alerter thread."""
if self._add_alert_thread.is_alive():
print("The alert thread is already started")
else:
Expand All @@ -71,65 +100,16 @@ def do_start_alerts(self, _: cmd2.Statement) -> None:
self._add_alert_thread.start()

def do_stop_alerts(self, _: cmd2.Statement) -> None:
"""Stops the alerter thread."""
"""Stop the alerter thread."""
self._stop_event.set()
if self._add_alert_thread.is_alive():
self._add_alert_thread.join()
else:
print("The alert thread is already stopped")

def _get_alerts(self) -> list[str]:
"""Reports alerts
:return: the list of alerts.
"""
cur_time = time.monotonic()
if cur_time < self._next_alert_time:
return []

alerts = []

if self._alert_count < len(ALERTS):
alerts.append(ALERTS[self._alert_count])
self._alert_count += 1
self._next_alert_time = cur_time + 4

else:
rand_num = self._secure_generator.randint(1, 20)
if rand_num > 2:
return []

for _ in range(rand_num):
self._alert_count += 1
alerts.append(f"Alert {self._alert_count}")

self._next_alert_time = 0

return alerts

def _build_alert_str(self) -> str:
"""Combines alerts into one string that can be printed to the terminal
:return: the alert string.
"""
alert_str = ""
alerts = self._get_alerts()

longest_alert = max(ALERTS, key=len)
num_asterisks = len(longest_alert) + 8

for i, cur_alert in enumerate(alerts):
# Use padding to center the alert
padding = " " * int((num_asterisks - len(cur_alert)) / 2)

if i > 0:
alert_str += "\n"
alert_str += "*" * num_asterisks + "\n"
alert_str += padding + cur_alert + padding + "\n"
alert_str += "*" * num_asterisks + "\n"

return alert_str

def _build_colored_prompt(self) -> str:
"""Randomly builds a colored prompt
"""Randomly build a colored prompt.

:return: the new prompt.
"""
rand_num = self._secure_generator.randint(1, 6)
Expand All @@ -150,28 +130,47 @@ def _build_colored_prompt(self) -> str:
return stylize(self.visible_prompt, style=status_color)

def _add_alerts_func(self) -> None:
"""Prints alerts and updates the prompt any time the prompt is showing."""
"""Print alerts and update the prompt any time the prompt is showing."""
self._alert_count = 0
self._next_alert_time = 0

alerts = get_alerts()
alert_index = 0
last_alert_time = 0.0

while not self._stop_event.is_set():
# Get any alerts that need to be printed
alert_str = self._build_alert_str()
cur_time = time.monotonic()
alert_msg = None
soft_wrap = True

# Trigger the next alert every 4 seconds
if cur_time - last_alert_time >= 4.0:
alert_msg, soft_wrap = alerts[alert_index]
alert_index = (alert_index + 1) % len(alerts)
self._alert_count += 1
last_alert_time = cur_time

# Build a new prompt
# Build a new prompt (color changes randomly)
new_prompt = self._build_colored_prompt()

# Check if we have alerts to print
if alert_str:
self.add_alert(msg=alert_str, prompt=new_prompt)
# Check if we have an alert to print
if alert_msg is not None:
# Wrap the alert message and an empty string in a Rich Group.
# This is a clean way to append a blank line after any RenderableType
# (strings, panels, tables, etc.) for visual separation.
self.add_alert(
msg=Group(alert_msg, ""),
soft_wrap=soft_wrap,
prompt=new_prompt,
)

new_title = f"Alerts Printed: {self._alert_count}"
self.set_window_title(new_title)

# Otherwise check if the prompt needs to be updated or refreshed
# Otherwise, check if the prompt needs to be updated or refreshed
elif self.prompt != new_prompt:
self.add_alert(prompt=new_prompt)

self._stop_event.wait(0.5)
self._stop_event.wait(1.0)


if __name__ == "__main__":
Expand Down
Loading
Loading