Execute the low-level input read from either a terminal or a redirected stream. 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
(
self,
prompt: Callable[[], ANSI | str] | ANSI | str,
session: PromptSession[str],
**prompt_kwargs: Any,
)
| 3433 | return not isinstance(session.input, DummyInput) |
| 3434 | |
| 3435 | def _read_raw_input( |
| 3436 | self, |
| 3437 | prompt: Callable[[], ANSI | str] | ANSI | str, |
| 3438 | session: PromptSession[str], |
| 3439 | **prompt_kwargs: Any, |
| 3440 | ) -> str: |
| 3441 | """Execute the low-level input read from either a terminal or a redirected stream. |
| 3442 | |
| 3443 | If input is coming from a TTY, it uses `prompt_toolkit` to render a |
| 3444 | UI with completion and `patch_stdout` protection. Otherwise it performs |
| 3445 | a direct line read from `stdin`. |
| 3446 | |
| 3447 | :param prompt: the prompt text or a callable that returns the prompt. |
| 3448 | :param session: the PromptSession instance to use for reading. |
| 3449 | :param prompt_kwargs: additional arguments passed directly to session.prompt(). |
| 3450 | :return: the stripped input string. |
| 3451 | :raises EOFError: if the input stream is closed or the user signals EOF (e.g., Ctrl+D) |
| 3452 | """ |
| 3453 | # Check if the session is configured for interactive terminal use. |
| 3454 | if self._is_tty_session(session): |
| 3455 | if not callable(prompt): |
| 3456 | prompt = pt_filter_style(prompt) |
| 3457 | |
| 3458 | with patch_stdout(): |
| 3459 | try: |
| 3460 | # Set this session as the active one for UI/completion logic. |
| 3461 | self.active_session = session |
| 3462 | return session.prompt(prompt, **prompt_kwargs) |
| 3463 | finally: |
| 3464 | # Revert back to the main session. |
| 3465 | self.active_session = self.main_session |
| 3466 | |
| 3467 | # We're not at a terminal, so we're likely reading from a file or a pipe. |
| 3468 | prompt_obj = prompt() if callable(prompt) else prompt |
| 3469 | prompt_str = prompt_obj.value if isinstance(prompt_obj, ANSI) else prompt_obj |
| 3470 | |
| 3471 | # If this is an interactive pipe, then display the prompt first |
| 3472 | if self.interactive_pipe: |
| 3473 | self.poutput(prompt_str, end="") |
| 3474 | self.stdout.flush() |
| 3475 | |
| 3476 | # Wait for the next line of input |
| 3477 | line = self.stdin.readline() |
| 3478 | |
| 3479 | # If the stream is empty, we've reached the end of the input. |
| 3480 | if not line: |
| 3481 | raise EOFError |
| 3482 | |
| 3483 | # If not interactive and echo is on, we want the output to simulate a |
| 3484 | # live session. Print the prompt and the command so they appear in the |
| 3485 | # output stream before the results. |
| 3486 | if not self.interactive_pipe and self.echo: |
| 3487 | end = "" if line.endswith("\n") else "\n" |
| 3488 | |
| 3489 | self.poutput(f"{prompt_str}{line}", end=end) |
| 3490 | |
| 3491 | return line.rstrip("\r\n") |
| 3492 |