Coverage for frappe_manager / output_manager / logging_output.py: 95%
100 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-07-02 18:13 +0530
1"""
2Logging output handler wrapper.
4This wrapper automatically logs all OutputHandler calls to a file logger,
5providing a single source of truth for debugging.
6"""
8import logging
9import sys
10from collections.abc import Iterable
11from typing import Any
13import typer
15from frappe_manager.logger.contextual import ContextualLogger
16from frappe_manager.output_manager.base import OutputHandler
19class LoggingOutputHandler(OutputHandler):
20 """
21 Output handler wrapper that logs all operations.
23 This handler wraps another OutputHandler and logs all method calls
24 to a Python logger, providing automatic logging of user-facing messages.
26 Supports both standard logging.Logger and ContextualLogger for automatic
27 context inclusion in log messages.
29 Usage:
30 # With standard logger
31 logger = get_logger()
32 rich = RichOutputHandler(verbose=True)
33 output = LoggingOutputHandler(rich, logger)
35 # With contextual logger
36 context = LoggerContext(bench="mybench", operation="create")
37 contextual_logger = ContextualLogger(logger, context)
38 output = LoggingOutputHandler(rich, contextual_logger)
40 # Now all output calls are automatically logged with context
41 output.print("Creating bench..")
42 # User sees: ⚡ Creating bench...
43 # Log file: [2024-01-07 12:00:00] INFO: [bench=mybench] [op=create] [OUTPUT] Creating bench...
44 """
46 def __init__(
47 self,
48 delegate: OutputHandler,
49 logger: logging.Logger | ContextualLogger,
50 log_prefix: str = "[OUTPUT]",
51 ):
52 """
53 Initialize logging wrapper.
55 Args:
56 delegate: OutputHandler to wrap and delegate to
57 logger: Python logger or ContextualLogger for file logging
58 log_prefix: Prefix for log messages (default: "[OUTPUT]")
59 """
60 super().__init__(delegate.verbose)
61 self.delegate = delegate
63 if isinstance(logger, logging.Logger):
64 self.logger = ContextualLogger(logger)
65 else:
66 self.logger = logger
68 self.log_prefix = log_prefix
70 def _log_message(self, level: int, message: str) -> None:
71 """
72 Helper to log with prefix.
74 Args:
75 level: Log level (logging.DEBUG, INFO, WARNING, ERROR)
76 message: Message to log
77 """
78 prefixed_message = f"{self.log_prefix} {message}"
80 if level == logging.DEBUG:
81 self.logger.debug(prefixed_message)
82 elif level == logging.INFO:
83 self.logger.info(prefixed_message)
84 elif level == logging.WARNING:
85 self.logger.warning(prefixed_message)
86 elif level == logging.ERROR:
87 self.logger.error(prefixed_message)
88 else:
89 self.logger.info(prefixed_message)
91 def start(self, text: str) -> None:
92 """
93 Start operation (logged at INFO level).
95 Args:
96 text: Initial status message
97 """
98 self._log_message(logging.INFO, f"START: {text}")
99 self.delegate.start(text)
100 super().start(text)
102 def change_head(self, text: str, style: str | None = None) -> None:
103 """
104 Change head (logged at DEBUG level).
106 Args:
107 text: New status message
108 style: Optional style hint
109 """
110 self._log_message(logging.DEBUG, f"CHANGE_HEAD: {text}")
111 self.delegate.change_head(text, style)
113 def update_head(self, text: str) -> None:
114 """
115 Update head (logged at DEBUG level).
117 Args:
118 text: New head text
119 """
120 self._log_message(logging.DEBUG, f"UPDATE_HEAD: {text}")
121 self.delegate.update_head(text)
123 def stop(self) -> None:
124 """
125 Stop operation (logged at DEBUG level).
126 """
127 self._log_message(logging.DEBUG, "STOP")
128 self.delegate.stop()
129 super().stop()
131 def debug(self, text: str, emoji_code: str = ":bug:", **kwargs) -> None:
132 """
133 Debug message (logged at DEBUG level).
135 Args:
136 text: Debug message
137 emoji_code: Optional emoji code
138 **kwargs: Additional arguments
139 """
140 self._log_message(logging.DEBUG, text)
141 self.delegate.debug(text, emoji_code, **kwargs)
143 def info(self, text: str, emoji_code: str = ":information:", **kwargs) -> None:
144 """
145 Info message (logged at INFO level).
147 Args:
148 text: Info message
149 emoji_code: Optional emoji code
150 **kwargs: Additional arguments
151 """
152 self._log_message(logging.INFO, text)
153 self.delegate.info(text, emoji_code, **kwargs)
155 def print(self, text: str, emoji_code: str = ":zap:", prefix: str | None = None, **kwargs) -> None:
156 """
157 Print message (logged at INFO level).
159 Args:
160 text: Message to print
161 emoji_code: Optional emoji code
162 prefix: Optional prefix
163 **kwargs: Additional arguments
164 """
165 full_text = f"{prefix} {text}" if prefix else text
166 self._log_message(logging.INFO, full_text)
167 self.delegate.print(text, emoji_code, prefix, **kwargs)
169 def display_error(self, text: str, emoji_code: str = ":no_entry:") -> None:
170 """
171 Display error (logged at ERROR level).
173 Args:
174 text: Error message
175 emoji_code: Optional emoji code
176 """
177 self._log_message(logging.ERROR, text)
178 self.delegate.display_error(text, emoji_code)
180 def error(self, text: str, exception: Exception, emoji_code: str = ":no_entry:") -> None:
181 """
182 Display error and raise the exception (logged at ERROR level with exception details).
184 This method always raises the provided exception after logging and displaying.
186 Args:
187 text: Error message
188 exception: Exception to raise (required)
189 emoji_code: Optional emoji code
191 Raises:
192 Exception: Always raises the provided exception
193 """
194 self._log_message(logging.ERROR, f"{text} | Exception: {type(exception).__name__}: {exception!s}")
195 self.logger.exception(f"{self.log_prefix} Exception details:", exc_info=exception)
196 self.delegate.error(text, exception, emoji_code)
198 def warning(self, text: str, emoji_code: str = ":warning:") -> None:
199 """
200 Display warning (logged at WARNING level).
202 Args:
203 text: Warning message
204 emoji_code: Optional emoji code
205 """
206 self._log_message(logging.WARNING, text)
207 self.delegate.warning(text, emoji_code)
209 def live_lines(
210 self,
211 data: Iterable[tuple[str, bytes]],
212 stdout: bool = True,
213 stderr: bool = True,
214 lines: int = 4,
215 padding: tuple[int, int, int, int] = (0, 0, 0, 2),
216 stop_string: str | None = None,
217 log_prefix: str = "=>",
218 ) -> None:
219 """
220 Display live lines (logs start/stop, not individual lines for performance).
222 Note: Individual lines are not logged to avoid performance overhead.
223 Only the start and completion of live output is logged.
225 Args:
226 data: Iterator yielding (source, line) tuples
227 stdout: Whether to display stdout lines
228 stderr: Whether to display stderr lines
229 lines: Maximum number of lines to display
230 padding: Padding around displayed lines
231 stop_string: String that stops display when found
232 log_prefix: Prefix for each line
233 """
234 self._log_message(logging.DEBUG, f"LIVE_LINES: Starting (stdout={stdout}, stderr={stderr})")
235 self.delegate.live_lines(data, stdout, stderr, lines, padding, stop_string, log_prefix)
236 self._log_message(logging.DEBUG, "LIVE_LINES: Completed")
238 def update_live(self, renderable: Any = None, padding: tuple[int, int, int, int] = (0, 0, 0, 0)) -> None:
239 """
240 Update live display (not logged for performance).
242 Args:
243 renderable: Content to display
244 padding: Padding around content
245 """
246 self.delegate.update_live(renderable, padding)
248 def prompt_ask(
249 self,
250 prompt: str = "",
251 choices: list | None = None,
252 default: str | None = None,
253 force_yes: bool = False,
254 required_flag: str | None = None,
255 **kwargs,
256 ) -> str:
257 self._log_message(logging.INFO, f"PROMPT: {prompt}")
259 response = self.delegate.prompt_ask(
260 prompt=prompt,
261 choices=choices,
262 default=default,
263 force_yes=force_yes,
264 required_flag=required_flag,
265 **kwargs,
266 )
268 if "password" not in prompt.lower():
269 self._log_message(logging.INFO, f"PROMPT_RESPONSE: {response}")
270 else:
271 self._log_message(logging.INFO, "PROMPT_RESPONSE: [password hidden]")
273 return response
275 def prompt_fuzzy(
276 self,
277 prompt: str,
278 choices: list[str],
279 default: str | None = None,
280 required_flag: str | None = None,
281 **kwargs,
282 ) -> str:
283 self.logger.info(f"[OUTPUT] FUZZY_PROMPT: {prompt} (choices: {len(choices)})")
285 result = self.delegate.prompt_fuzzy(
286 prompt=prompt,
287 choices=choices,
288 default=default,
289 required_flag=required_flag,
290 **kwargs,
291 )
293 self.logger.info(f"[OUTPUT] FUZZY_RESPONSE: {result}")
294 return result
296 def set_interactive_mode(self, non_interactive_flag: bool) -> None:
297 """
298 Set interactive mode on wrapper AND delegate.
300 This ensures the delegate (RichOutputHandler) respects the non-interactive flag
301 when deciding whether to show spinners.
303 Args:
304 non_interactive_flag: True to disable interactive features (spinners, prompts)
305 """
306 # Set on wrapper itself
307 super().set_interactive_mode(non_interactive_flag)
309 # Forward to delegate so it respects the flag too
310 if hasattr(self.delegate, "set_interactive_mode"):
311 self.delegate.set_interactive_mode(non_interactive_flag)
313 @property
314 def should_stream_docker(self) -> bool:
315 return self.delegate.should_stream_docker
317 def print_data(self, data: Any, **kwargs) -> None:
318 self._log_message(logging.INFO, f"DATA: {data}")
319 self.delegate.print_data(data, **kwargs)
321 def print_status(self, text: str, emoji_code: str = ":zap:", **kwargs) -> None:
322 self._log_message(logging.INFO, f"STATUS: {text}")
323 self.delegate.print_status(text, emoji_code, **kwargs)
325 def exit(self, text: str, emoji_code: str = ":no_entry:", os_exit=False, error_msg=None):
326 self._log_message(logging.ERROR, f"EXIT: {text}" + (f" | Error: {error_msg}" if error_msg else ""))
327 exit_method = getattr(self.delegate, "exit", None)
328 if exit_method and callable(exit_method):
329 exit_method(text, emoji_code, os_exit, error_msg)
330 else:
331 self.display_error(text, emoji_code)
332 if os_exit:
333 sys.exit(1)
334 raise typer.Exit(1)