Read the next command line from the input stream. :param prompt: prompt to display to user :return: the line read from stdin with all trailing new lines removed :raises EOFError: if the input stream is closed or the user signals EOF (e.g., Ctrl+D) :raises Exception:
(self, prompt: str)
| 3639 | get_app().invalidate() |
| 3640 | |
| 3641 | def _read_command_line(self, prompt: str) -> str: |
| 3642 | """Read the next command line from the input stream. |
| 3643 | |
| 3644 | :param prompt: prompt to display to user |
| 3645 | :return: the line read from stdin with all trailing new lines removed |
| 3646 | :raises EOFError: if the input stream is closed or the user signals EOF (e.g., Ctrl+D) |
| 3647 | :raises Exception: any other exceptions raised by prompt() |
| 3648 | """ |
| 3649 | |
| 3650 | # Use dynamic prompt if the prompt matches self.prompt |
| 3651 | def get_prompt() -> str | ANSI: |
| 3652 | return pt_filter_style(self.prompt) |
| 3653 | |
| 3654 | prompt_to_use: Callable[[], ANSI | str] | ANSI | str = ANSI(prompt) |
| 3655 | if prompt == self.prompt: |
| 3656 | prompt_to_use = get_prompt |
| 3657 | |
| 3658 | def _pre_prompt() -> None: |
| 3659 | """Run standard pre-prompt processing and activate the background alerter.""" |
| 3660 | # Record prompt start time so any async prompt updates queued during |
| 3661 | # pre_prompt() are considered current. |
| 3662 | self._alert_prompt_timestamp = time.monotonic() |
| 3663 | self.pre_prompt() |
| 3664 | |
| 3665 | # Start alerter thread if it's not already running. |
| 3666 | if self._alert_thread is None or not self._alert_thread.is_alive(): |
| 3667 | self._alert_allowed = False |
| 3668 | self._alert_shutdown = False |
| 3669 | self._alert_thread = threading.Thread(target=self._process_alerts, daemon=True) |
| 3670 | self._alert_thread.start() |
| 3671 | |
| 3672 | # Allow alerts to be printed now that we are at a prompt. |
| 3673 | with self._alert_condition: |
| 3674 | self._alert_allowed = True |
| 3675 | self._alert_condition.notify_all() |
| 3676 | |
| 3677 | try: |
| 3678 | return self._read_raw_input( |
| 3679 | prompt=prompt_to_use, |
| 3680 | session=self.main_session, |
| 3681 | pre_run=_pre_prompt, |
| 3682 | ) |
| 3683 | finally: |
| 3684 | # Ensure no alerts print while not at a prompt. |
| 3685 | with self._alert_condition: |
| 3686 | self._alert_allowed = False |
| 3687 | |
| 3688 | def _cmdloop(self) -> None: |
| 3689 | """Repeatedly issue a prompt, accept input, parse it, and dispatch to appropriate commands. |