Coverage for frappe_manager / output_manager / base.py: 98%
60 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-07-02 18:13 +0530
1"""
2Base interface for output handling.
4This abstract base class defines the contract that all output handlers must implement,
5allowing business logic to be independent of the presentation layer.
6"""
8import sys
9from abc import ABC, abstractmethod
10from collections.abc import Iterable
11from typing import Any
14class OutputHandler(ABC):
15 """
16 Abstract base class for handling output in business logic.
18 This interface provides a unified API for different output mechanisms
19 (CLI, API, testing, etc.) without coupling business logic to any specific
20 presentation layer.
21 """
23 def __init__(self, verbose: bool = False):
24 """
25 Initialize output handler with verbosity settings.
27 Args:
28 verbose: Show info and debug level messages
29 """
30 self.verbose = verbose
31 self._spinner_active = False
32 self._current_text: str | None = None
34 # Interactive mode state (3-level priority system)
35 self._interactive: bool | None = None # None = not initialized
36 self._tty_available: bool = sys.stdin.isatty() and sys.stdout.isatty()
38 @abstractmethod
39 def start(self, text: str) -> None:
40 """
41 Start a new operation with a status message.
43 Args:
44 text: The initial status message to display
45 """
46 self._spinner_active = True
47 self._current_text = text
49 @abstractmethod
50 def change_head(self, text: str, style: str | None = None) -> None:
51 """
52 Update the current operation status message.
54 Args:
55 text: The new status message
56 style: Optional style hint (implementation-specific)
57 """
59 @abstractmethod
60 def update_head(self, text: str) -> None:
61 """
62 Update the head text (similar to change_head but with different semantics).
64 Args:
65 text: The new head text
66 """
68 @abstractmethod
69 def stop(self) -> None:
70 """
71 Stop the current operation status display.
72 """
73 self._spinner_active = False
75 @abstractmethod
76 def print(self, text: str, emoji_code: str = ":zap:", prefix: str | None = None, **kwargs) -> None:
77 """
78 Print a message.
80 Args:
81 text: The message to print
82 emoji_code: Optional emoji code (implementation-specific)
83 prefix: Optional prefix for the message
84 **kwargs: Additional implementation-specific arguments
85 """
87 @abstractmethod
88 def debug(self, text: str, emoji_code: str = ":bug:", **kwargs) -> None:
89 """
90 Display debug message (only shown if verbose=True).
92 Args:
93 text: Debug message
94 emoji_code: Optional emoji code
95 **kwargs: Additional implementation-specific arguments
96 """
98 @abstractmethod
99 def info(self, text: str, emoji_code: str = ":information:", **kwargs) -> None:
100 """
101 Display info message (only shown if verbose=True).
103 Args:
104 text: Info message
105 emoji_code: Optional emoji code
106 **kwargs: Additional implementation-specific arguments
107 """
109 @abstractmethod
110 def display_error(self, text: str, emoji_code: str = ":no_entry:") -> None:
111 """
112 Display error message without raising exception.
114 Args:
115 text: Error message
116 emoji_code: Optional emoji code
117 """
119 @abstractmethod
120 def error(self, text: str, exception: Exception, emoji_code: str = ":no_entry:") -> None:
121 """
122 Display an error message and raise the exception.
124 This method always raises the provided exception after displaying the error message.
125 Use display_error() if you want to display an error without raising an exception.
127 Args:
128 text: The error message
129 exception: Exception to raise after displaying (required)
130 emoji_code: Optional emoji code (implementation-specific)
132 Raises:
133 Exception: Always raises the provided exception
134 """
136 @abstractmethod
137 def warning(self, text: str, emoji_code: str = ":warning:") -> None:
138 """
139 Display a warning message.
141 Args:
142 text: The warning message
143 emoji_code: Optional emoji code (implementation-specific)
144 """
146 @abstractmethod
147 def live_lines(
148 self,
149 data: Iterable[tuple[str, bytes]],
150 stdout: bool = True,
151 stderr: bool = True,
152 lines: int = 4,
153 padding: tuple[int, int, int, int] = (0, 0, 0, 2),
154 stop_string: str | None = None,
155 log_prefix: str = "=>",
156 ) -> None:
157 """
158 Display live streaming output from a process.
160 Args:
161 data: Iterable yielding (source, line) tuples where source is "stdout" or "stderr"
162 stdout: Whether to display stdout lines
163 stderr: Whether to display stderr lines
164 lines: Maximum number of lines to display
165 padding: Padding around displayed lines (top, right, bottom, left)
166 stop_string: String that stops display when found
167 log_prefix: Prefix for each line
168 """
170 @abstractmethod
171 def update_live(self, renderable: Any = None, padding: tuple[int, int, int, int] = (0, 0, 0, 0)) -> None:
172 """
173 Update the live display with new content.
175 Args:
176 renderable: Content to display (implementation-specific)
177 padding: Padding around content (top, right, bottom, left)
178 """
180 def set_interactive_mode(self, non_interactive_flag: bool) -> None:
181 """
182 Set interactive mode from global --non-interactive flag and TTY detection.
184 This implements priority level 2 & 3 of the interactive mode system:
185 - Priority 1 (command flags like --force) handled in prompt_ask()
186 - Priority 2: Global --non-interactive flag (this method)
187 - Priority 3: Auto-detect TTY (fallback)
189 Args:
190 non_interactive_flag: True if --non-interactive was passed
191 """
192 if non_interactive_flag:
193 self._interactive = False
194 else:
195 self._interactive = self._tty_available
197 def is_interactive(self) -> bool:
198 """
199 Check if prompts should be shown.
201 Returns:
202 True if interactive mode is enabled (prompts allowed)
203 """
204 if self._interactive is None:
205 return self._tty_available
206 return self._interactive
208 @abstractmethod
209 def prompt_ask(
210 self,
211 prompt: str = "",
212 choices: list | None = None,
213 default: str | None = None,
214 force_yes: bool = False,
215 required_flag: str | None = None,
216 **kwargs,
217 ) -> str:
218 """
219 Prompt the user for input with 3-level priority system.
221 Priority system:
222 1. If force_yes=True → return "yes" (command-specific flags like --force)
223 2. If not interactive → return default or raise NonInteractiveError
224 3. Else → show prompt (interactive mode)
226 Args:
227 prompt: Question to ask the user
228 choices: List of valid choices (optional)
229 default: Default value if non-interactive (required for non-interactive)
230 force_yes: Force "yes" response (for command-specific --force/--yes flags)
231 required_flag: Flag name to suggest in error message (e.g., "--yes")
232 If set and non-interactive, raises NonInteractiveError with this suggestion
233 **kwargs: Implementation-specific prompt arguments
235 Returns:
236 The user's input as a string
238 Raises:
239 NonInteractiveError: If non-interactive and no default provided, or if required_flag is set
240 """
242 @abstractmethod
243 def prompt_fuzzy(
244 self,
245 prompt: str,
246 choices: list[str],
247 default: str | None = None,
248 required_flag: str | None = None,
249 **kwargs,
250 ) -> str:
251 """
252 Prompt the user with fuzzy search selection (respects interactive mode).
254 Uses same priority system as prompt_ask():
255 1. If not interactive → return default or raise NonInteractiveError
256 2. Else → show fuzzy search prompt (interactive mode)
258 Args:
259 prompt: Message to display to user
260 choices: List of options for fuzzy search
261 default: Default value if non-interactive (optional)
262 required_flag: Flag name to suggest in error message (e.g., "bench_name positional argument")
263 If set and non-interactive, raises NonInteractiveError with this suggestion
264 **kwargs: Implementation-specific fuzzy prompt arguments (vi_mode, qmark, etc.)
266 Returns:
267 Selected choice as string
269 Raises:
270 NonInteractiveError: If non-interactive and no default provided, or if required_flag is set
271 """
273 @property
274 def is_spinner_active(self) -> bool:
275 return self._spinner_active
277 @property
278 @abstractmethod
279 def should_stream_docker(self) -> bool:
280 """
281 Determine if docker output should be streamed based on output context.
283 Returns True when docker operations should stream their output (e.g., spinner active in TTY),
284 False when operations should run quietly with no intermediate output.
286 Returns:
287 bool: True to stream docker output, False to suppress it
288 """
290 @abstractmethod
291 def print_data(self, data: Any, **kwargs) -> None:
292 """
293 Print structured data to stdout (pipeable).
295 Use for data that users want to pipe or process:
296 - Tables (fm list)
297 - JSON output
298 - Query results
300 Args:
301 data: Data to print
302 **kwargs: Format-specific options
303 """
305 @abstractmethod
306 def print_status(self, text: str, emoji_code: str = ":zap:", **kwargs) -> None:
307 """
308 Print status/diagnostic message to stderr.
310 Use for:
311 - Progress messages
312 - Warnings
313 - Errors
314 - Success confirmations
316 Args:
317 text: Status message
318 emoji_code: Emoji code
319 **kwargs: Additional arguments
320 """