Coverage for frappe_manager / output_manager / json_output.py: 85%
106 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-07-02 18:13 +0530
1"""
2JSON event output handler.
4This implementation captures all output events as structured JSON data,
5suitable for API responses, logging, or testing. No actual output is
6displayed to the terminal.
7"""
9import json
10from collections.abc import Iterable
11from typing import Any
13from frappe_manager.output_manager.base import OutputHandler
16class OutputEvent:
17 """Represents a single output event."""
19 def __init__(self, event_type: str, data: dict):
20 """
21 Initialize an output event.
23 Args:
24 event_type: Type of event (e.g., "start", "print", "error")
25 data: Event-specific data
26 """
27 self.event_type = event_type
28 self.data = data
30 def to_dict(self) -> dict:
31 """
32 Convert event to dictionary.
34 Returns:
35 Dictionary representation of the event
36 """
37 return {"event_type": self.event_type, "data": self.data}
39 def to_json(self) -> str:
40 """
41 Convert event to JSON string.
43 Returns:
44 JSON string representation of the event
45 """
46 return json.dumps(self.to_dict())
49class JSONOutputHandler(OutputHandler):
50 """
51 Output handler that captures events as structured JSON data.
53 This handler is suitable for:
54 - API responses (FastAPI, Flask, etc.)
55 - Structured logging
56 - Testing and assertions
57 - WebSocket communication
59 All events are captured in the `events` list and can be retrieved
60 as dictionaries or JSON strings.
61 """
63 def __init__(self, verbose: bool = False, persist_to_file: Any = None):
64 """
65 Initialize the JSON output handler.
67 Args:
68 verbose: Capture info and debug level messages
69 persist_to_file: Optional file path to persist events (JSONL format)
70 """
71 super().__init__(verbose)
72 self.events: list[OutputEvent] = []
73 self._current_head: str | None = None
74 self._is_started: bool = False
75 self.persist_file = persist_to_file
77 def _add_event(self, event: OutputEvent) -> None:
78 """
79 Add event to list and optionally persist to file.
81 Args:
82 event: OutputEvent to add
83 """
84 self.events.append(event)
86 if self.persist_file:
87 with open(self.persist_file, "a") as f:
88 f.write(event.to_json() + "\n")
90 def start(self, text: str) -> None:
91 """
92 Start a new operation with a status message.
94 Args:
95 text: The initial status message
96 """
97 super().start(text)
98 self._current_head = text
99 self._is_started = True
100 self._add_event(OutputEvent("start", {"text": text}))
102 def change_head(self, text: str, style: str | None = None) -> None:
103 """
104 Update the current operation status message.
106 Args:
107 text: The new status message
108 style: Optional style hint (ignored in JSON output)
109 """
110 previous_head = self._current_head
111 self._current_head = text
112 self._add_event(
113 OutputEvent("change_head", {"text": text, "previous": previous_head, "style": style}),
114 )
116 def update_head(self, text: str) -> None:
117 """
118 Update the head text.
120 Args:
121 text: The new head text
122 """
123 previous_head = self._current_head
124 self._current_head = text
125 self._add_event(OutputEvent("update_head", {"text": text, "previous": previous_head}))
127 def stop(self) -> None:
128 """
129 Stop the current operation status display.
130 """
131 super().stop()
132 self._is_started = False
133 self._add_event(OutputEvent("stop", {}))
135 def print(self, text: str, emoji_code: str = ":zap:", prefix: str | None = None, **kwargs) -> None:
136 """
137 Print a message.
139 Args:
140 text: The message to print
141 emoji_code: Emoji code (captured but not rendered)
142 prefix: Optional prefix for the message
143 **kwargs: Additional arguments (captured in data)
144 """
145 self._add_event(
146 OutputEvent(
147 "print",
148 {"text": text, "emoji_code": emoji_code, "prefix": prefix, "kwargs": kwargs},
149 ),
150 )
152 def debug(self, text: str, emoji_code: str = ":bug:", **kwargs) -> None:
153 """
154 Capture debug message if debug mode is enabled.
156 Args:
157 text: Debug message
158 emoji_code: Emoji code (captured but not rendered)
159 **kwargs: Additional arguments (captured in data)
160 """
161 if self.verbose:
162 self._add_event(OutputEvent("debug", {"text": text, "emoji_code": emoji_code, "kwargs": kwargs}))
164 def info(self, text: str, emoji_code: str = ":information:", **kwargs) -> None:
165 """
166 Capture info message if verbose mode is enabled.
168 Args:
169 text: Info message
170 emoji_code: Emoji code (captured but not rendered)
171 **kwargs: Additional arguments (captured in data)
172 """
173 if self.verbose:
174 self._add_event(OutputEvent("info", {"text": text, "emoji_code": emoji_code, "kwargs": kwargs}))
176 def display_error(self, text: str, emoji_code: str = ":no_entry:") -> None:
177 """
178 Capture error message without raising exception.
180 Args:
181 text: Error message
182 emoji_code: Emoji code (captured but not rendered)
183 """
184 self._add_event(OutputEvent("display_error", {"text": text, "emoji_code": emoji_code}))
186 def error(self, text: str, exception: Exception, emoji_code: str = ":no_entry:") -> None:
187 """
188 Display an error message and raise the exception.
190 This method always raises the provided exception after capturing the event.
192 Args:
193 text: The error message
194 exception: Exception to raise after capturing (required)
195 emoji_code: Emoji code (captured but not rendered)
197 Raises:
198 Exception: Always raises the provided exception
199 """
200 self._add_event(
201 OutputEvent(
202 "error",
203 {
204 "text": text,
205 "emoji_code": emoji_code,
206 "exception": str(exception),
207 "exception_type": type(exception).__name__,
208 },
209 ),
210 )
211 raise exception
213 def warning(self, text: str, emoji_code: str = ":warning:") -> None:
214 """
215 Display a warning message.
217 Args:
218 text: The warning message
219 emoji_code: Emoji code (captured but not rendered)
220 """
221 self._add_event(OutputEvent("warning", {"text": text, "emoji_code": emoji_code}))
223 def live_lines(
224 self,
225 data: Iterable[tuple[str, bytes]],
226 stdout: bool = True,
227 stderr: bool = True,
228 lines: int = 4,
229 padding: tuple[int, int, int, int] = (0, 0, 0, 2),
230 stop_string: str | None = None,
231 log_prefix: str = "=>",
232 ) -> None:
233 """
234 Display live streaming output from a process.
236 Args:
237 data: Iterator yielding (source, line) tuples
238 stdout: Whether to capture stdout lines
239 stderr: Whether to capture stderr lines
240 lines: Maximum number of lines (hint only)
241 padding: Padding (ignored in JSON output)
242 stop_string: String that stops capture when found
243 log_prefix: Prefix for each line
244 """
245 captured_lines: list[dict] = []
247 for source, line in data:
248 try:
249 decoded_line = line.decode()
250 except Exception:
251 decoded_line = str(line)
253 if source == "stdout" and not stdout:
254 continue
255 if source == "stderr" and not stderr:
256 continue
258 captured_lines.append({"source": source, "line": decoded_line})
260 if stop_string and stop_string.lower() in decoded_line.lower():
261 break
263 self._add_event(
264 OutputEvent(
265 "live_lines",
266 {
267 "lines": captured_lines,
268 "stdout": stdout,
269 "stderr": stderr,
270 "max_lines": lines,
271 "stop_string": stop_string,
272 "log_prefix": log_prefix,
273 },
274 ),
275 )
277 def update_live(self, renderable: Any = None, padding: tuple[int, int, int, int] = (0, 0, 0, 0)) -> None:
278 """
279 Update the live display with new content.
281 Args:
282 renderable: Content to display (captured as string)
283 padding: Padding (ignored in JSON output)
284 """
285 self._add_event(
286 OutputEvent("update_live", {"renderable": str(renderable) if renderable else None, "padding": padding}),
287 )
289 def prompt_ask(
290 self,
291 prompt: str = "",
292 choices: list | None = None,
293 default: str | None = None,
294 force_yes: bool = False,
295 required_flag: str | None = None,
296 **kwargs,
297 ) -> str:
298 from frappe_manager.exceptions import NonInteractiveError
300 self._add_event(
301 OutputEvent(
302 "prompt_ask",
303 {
304 "prompt": prompt,
305 "choices": choices,
306 "default": default,
307 "force_yes": force_yes,
308 "required_flag": required_flag,
309 "kwargs": kwargs,
310 },
311 ),
312 )
314 if force_yes:
315 return "yes"
317 if required_flag:
318 raise NonInteractiveError(
319 f"Cannot prompt in JSON output mode: {prompt}",
320 suggestions=[f"Provide: {required_flag}"],
321 )
323 if default is not None:
324 return default
326 raise NonInteractiveError(
327 f"Cannot prompt in JSON output mode: {prompt}",
328 suggestions=["Use interactive mode for prompts"],
329 )
331 def prompt_fuzzy(
332 self,
333 prompt: str,
334 choices: list[str],
335 default: str | None = None,
336 required_flag: str | None = None,
337 **kwargs,
338 ) -> str:
339 from frappe_manager.exceptions import NonInteractiveError
341 self._add_event(
342 OutputEvent(
343 "prompt_fuzzy",
344 {
345 "prompt": prompt,
346 "choices": choices,
347 "default": default,
348 "required_flag": required_flag,
349 "kwargs": kwargs,
350 },
351 ),
352 )
354 if required_flag:
355 raise NonInteractiveError(
356 f"Cannot prompt in JSON output mode: {prompt}",
357 suggestions=[f"Provide: {required_flag}"],
358 )
360 if default is not None:
361 return default
363 raise NonInteractiveError(
364 f"Cannot prompt in JSON output mode: {prompt}",
365 suggestions=["Use interactive mode for prompts"],
366 )
368 @property
369 def should_stream_docker(self) -> bool:
370 return False
372 def print_data(self, data: Any, **kwargs) -> None:
373 self._add_event(OutputEvent("print_data", {"data": data, "kwargs": kwargs}))
375 def print_status(self, text: str, emoji_code: str = ":zap:", **kwargs) -> None:
376 self._add_event(OutputEvent("print_status", {"text": text, "emoji_code": emoji_code, "kwargs": kwargs}))
378 def get_events(self) -> list[dict]:
379 """
380 Get all captured events as dictionaries.
382 Returns:
383 List of event dictionaries
384 """
385 return [event.to_dict() for event in self.events]
387 def get_events_json(self) -> str:
388 """
389 Get all captured events as JSON string.
391 Returns:
392 JSON string containing all events
393 """
394 return json.dumps(self.get_events(), indent=2)
396 def clear_events(self) -> None:
397 """
398 Clear all captured events.
399 """
400 self.events.clear()
401 self._current_head = None
402 self._is_started = False