Coverage for cli / output.py: 100.00%
398 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""
2Output formatting for GCO CLI.
4Provides consistent output formatting across all CLI commands
5with support for table, JSON, and YAML formats.
6"""
8import json
9import os
10import sys
11import tempfile
12import threading
13from collections.abc import Sequence
14from contextlib import redirect_stdout
15from contextvars import ContextVar
16from dataclasses import asdict, is_dataclass
17from datetime import datetime
18from io import StringIO
19from typing import Any
21import click
22import yaml
24from .config import GCOConfig, get_config
27def _serialize_value(value: Any) -> Any:
28 """Serialize a value for output."""
29 if isinstance(value, datetime):
30 return value.isoformat()
31 if is_dataclass(value) and not isinstance(value, type):
32 return asdict(value)
33 if isinstance(value, dict):
34 return {k: _serialize_value(v) for k, v in value.items()}
35 if isinstance(value, list):
36 return [_serialize_value(v) for v in value]
37 return value
40_structured_emissions_var: ContextVar[list[tuple[Any, str]] | None] = ContextVar(
41 "gco_structured_emissions", default=None
42)
43_structured_exit_code_var: ContextVar[int | None] = ContextVar(
44 "gco_structured_exit_code", default=None
45)
46_STDOUT_CAPTURE_LOCK = threading.RLock()
49def _reject_nonstandard_json_constant(token: str) -> None:
50 raise ValueError(f"non-standard JSON constant: {token}")
53def _machine_output_active() -> bool:
54 """Return whether the active root or direct group uses JSON/YAML output."""
55 if _structured_emissions_var.get() is not None:
56 return True
58 context = click.get_current_context(silent=True)
59 while context is not None:
60 output_format = context.params.get("output_format")
61 if output_format in {"json", "yaml"}:
62 return True
63 configured_format = getattr(context.obj, "output_format", None)
64 if configured_format in {"json", "yaml"}:
65 return True
66 context = context.parent
67 return False
70def confirm(message: str, *args: Any, **kwargs: Any) -> bool:
71 """Run a Click confirmation without contaminating machine stdout."""
72 if "err" not in kwargs and _machine_output_active():
73 kwargs["err"] = True
74 return click.confirm(message, *args, **kwargs)
77def prompt(message: str, *args: Any, **kwargs: Any) -> Any:
78 """Run a Click prompt without contaminating machine stdout."""
79 if "err" not in kwargs and _machine_output_active():
80 kwargs["err"] = True
81 return click.prompt(message, *args, **kwargs)
84def interactive_echo(message: Any = None, **kwargs: Any) -> None:
85 """Emit interactive context on stderr in machine-readable modes."""
86 if "err" not in kwargs and _machine_output_active():
87 kwargs["err"] = True
88 click.echo(message, **kwargs)
91def _canonical_emission_text(text: str) -> str:
92 """Normalize platform line endings only for native-emission matching."""
93 return text.replace("\r\n", "\n").replace("\r", "\n")
96def _structured_document_from_text(
97 text: str,
98 output_format: str,
99 emissions: Sequence[tuple[Any, str]] = (),
100) -> Any:
101 """Normalize one command's successful stdout into one structured value.
103 A single document emitted through :func:`emit_structured_document` keeps
104 its native mapping/list/null schema (or receives the scalar envelope).
105 Unregistered stdout is always raw text: YAML accepts most prose as valid
106 syntax, so parsing captured bytes cannot reliably distinguish a command
107 payload from logs, help text, or streamed output.
108 """
109 del output_format # The requested format controls rendering, not classification.
110 stripped = text.strip()
111 if not stripped:
112 return {"status": "ok"}
114 if len(emissions) == 1:
115 document, emitted_text = emissions[0]
116 if _canonical_emission_text(text) == _canonical_emission_text(emitted_text):
117 if isinstance(document, (dict, list)) or document is None:
118 return document
119 return {"status": "ok", "result": document}
121 return {"status": "ok", "output": text.rstrip("\r\n")}
124def _render_structured_document(document: Any, output_format: str) -> str:
125 """Render a normalized document in the exact root-requested format."""
126 serialized = _serialize_value(document)
127 if output_format == "json":
128 return json.dumps(serialized, indent=2, default=str, allow_nan=False)
129 return str(yaml.safe_dump(serialized, default_flow_style=False, sort_keys=False)).rstrip("\n")
132def emit_structured_document(
133 document: Any,
134 *,
135 output_format: str,
136 rendered: str | None = None,
137 err: bool = False,
138 nl: bool = True,
139) -> None:
140 """Emit and register one command-native machine document.
142 ``rendered`` lets legacy JSON-only command surfaces preserve their exact
143 bytes while still identifying the underlying payload to the root output
144 transaction. Error documents stay on stderr and are never registered as a
145 successful stdout payload.
146 """
147 output = (
148 rendered if rendered is not None else _render_structured_document(document, output_format)
149 )
150 emitted_text = f"{output}\n" if nl else output
151 emissions = _structured_emissions_var.get()
152 if not err and emissions is not None:
153 registered_document = _serialize_value(document)
154 if output_format == "json":
155 registered_document = json.loads(
156 output,
157 parse_constant=_reject_nonstandard_json_constant,
158 )
159 emissions.append((registered_document, emitted_text))
160 click.echo(output, err=err, nl=nl)
163def _write_structured_capture(
164 text: str,
165 output_format: str,
166 emissions: Sequence[tuple[Any, str]],
167) -> None:
168 rendered = _render_structured_document(
169 _structured_document_from_text(text, output_format, emissions),
170 output_format,
171 )
172 sys.stdout.write(f"{rendered}\n")
175def _root_output_settings(args: Sequence[str]) -> tuple[str | None, str | None]:
176 """Read root output/config values with Click-compatible short clusters."""
177 output_format: str | None = None
178 config_file: str | None = None
179 index = 0
180 while index < len(args):
181 arg = args[index]
182 if arg == "--" or not arg.startswith("-"):
183 break
185 if arg in {"--output", "--config", "--region"}:
186 value = args[index + 1] if index + 1 < len(args) else None
187 if arg == "--output":
188 output_format = value
189 elif arg == "--config":
190 config_file = value
191 index += 2
192 continue
193 if arg.startswith("--output="):
194 output_format = arg.partition("=")[2]
195 index += 1
196 continue
197 if arg.startswith("--config="):
198 config_file = arg.partition("=")[2]
199 index += 1
200 continue
201 if arg.startswith("--"):
202 index += 1
203 continue
205 cluster = arg[1:]
206 position = 0
207 consumed_next = False
208 while position < len(cluster):
209 option = cluster[position]
210 if option == "v":
211 position += 1
212 continue
213 if option not in {"o", "c", "r"}:
214 break
216 attached_value = cluster[position + 1 :].removeprefix("=")
217 value = attached_value or (args[index + 1] if index + 1 < len(args) else None)
218 consumed_next = not attached_value and value is not None
219 if option == "o":
220 output_format = value
221 elif option == "c":
222 config_file = value
223 # A value-taking short option consumes the rest of its cluster.
224 break
225 index += 2 if consumed_next else 1
226 return output_format, config_file
229def _requested_output_format(args: Sequence[str]) -> str | None:
230 """Resolve the root output mode early enough to include eager options."""
231 output_format, config_file = _root_output_settings(args)
232 if output_format is not None:
233 return output_format
234 try:
235 return get_config(config_file).output_format
236 except Exception:
237 # Let Click and the root callback surface the original config failure.
238 return None
241def _shell_completion_requested(complete_var: str | None, prog_name: str | None) -> bool:
242 """Keep Click's shell-completion protocol outside output normalization."""
243 if complete_var is None:
244 detected_name = prog_name or os.path.basename(sys.argv[0])
245 complete_name = detected_name.replace("-", "_").replace(".", "_")
246 complete_var = f"_{complete_name}_COMPLETE".upper()
247 return bool(os.environ.get(complete_var))
250class _StdoutCapture:
251 """Capture Python and inherited child-process stdout for one invocation."""
253 def __init__(self) -> None:
254 self.text = ""
255 self.raw_bytes: bytes | None = None
256 self._stream: Any = None
257 self._stream_encoding = "utf-8"
258 self._stream_fd: int | None = None
259 self._saved_fd: int | None = None
260 self._temporary: Any = None
261 self._string_buffer: StringIO | None = None
262 self._redirect: Any = None
263 self._owns_lock = False
265 def _release_lock(self) -> None:
266 if self._owns_lock:
267 self._owns_lock = False
268 _STDOUT_CAPTURE_LOCK.release()
270 def replay(self) -> None:
271 """Replay captured output without applying text newline translation twice."""
272 if self.raw_bytes is None or self._stream_fd is None:
273 sys.stdout.write(self.text)
274 return
276 self._stream.flush()
277 remaining = memoryview(self.raw_bytes)
278 while remaining:
279 written = os.write(self._stream_fd, remaining)
280 if written <= 0:
281 raise OSError("stdout replay made no progress")
282 remaining = remaining[written:]
284 def __enter__(self) -> _StdoutCapture:
285 _STDOUT_CAPTURE_LOCK.acquire()
286 self._owns_lock = True
287 self._stream = sys.stdout
288 self._stream_encoding = getattr(self._stream, "encoding", None) or "utf-8"
289 try:
290 try:
291 self._stream_fd = self._stream.fileno()
292 self._stream.flush()
293 self._saved_fd = os.dup(self._stream_fd)
294 self._temporary = tempfile.TemporaryFile(mode="w+b")
295 os.dup2(self._temporary.fileno(), self._stream_fd)
296 except AttributeError, OSError, ValueError:
297 if self._saved_fd is not None:
298 os.close(self._saved_fd)
299 self._saved_fd = None
300 if self._temporary is not None:
301 self._temporary.close()
302 self._temporary = None
303 self._stream_fd = None
304 self._string_buffer = StringIO()
305 self._redirect = redirect_stdout(self._string_buffer)
306 self._redirect.__enter__()
307 return self
308 except BaseException:
309 self._release_lock()
310 raise
312 def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None:
313 cleanup_error: Exception | None = None
315 def remember(error: Exception) -> None:
316 nonlocal cleanup_error
317 if cleanup_error is None:
318 cleanup_error = error
320 try:
321 if self._redirect is not None:
322 try:
323 self._redirect.__exit__(exc_type, exc, traceback)
324 except Exception as error:
325 remember(error)
326 try:
327 assert self._string_buffer is not None
328 self.text = self._string_buffer.getvalue()
329 except Exception as error:
330 remember(error)
331 else:
332 assert self._stream_fd is not None
333 assert self._saved_fd is not None
334 assert self._temporary is not None
335 try:
336 self._stream.flush()
337 except Exception as error:
338 remember(error)
339 restored = False
340 try:
341 os.dup2(self._saved_fd, self._stream_fd)
342 restored = True
343 except Exception:
344 # Keep the original descriptor alive and retry once before
345 # giving up; a transient restore failure must not strand
346 # process stdout on the capture file.
347 try:
348 os.dup2(self._saved_fd, self._stream_fd)
349 restored = True
350 except Exception as error:
351 remember(error)
352 if restored:
353 try:
354 os.close(self._saved_fd)
355 except Exception as error:
356 remember(error)
357 finally:
358 self._saved_fd = None
359 try:
360 self._temporary.seek(0)
361 captured_bytes = self._temporary.read()
362 self.raw_bytes = captured_bytes
363 try:
364 self.text = captured_bytes.decode(
365 self._stream_encoding,
366 errors="replace",
367 )
368 except LookupError:
369 self.text = captured_bytes.decode("utf-8", errors="replace")
370 except Exception as error:
371 remember(error)
372 finally:
373 if restored:
374 try:
375 self._temporary.close()
376 except Exception as error:
377 remember(error)
378 self._temporary = None
379 finally:
380 self._release_lock()
382 if cleanup_error is not None and exc_type is None:
383 raise cleanup_error
386class StructuredOutputGroup(click.Group):
387 """Root group that commits exactly one document for JSON/YAML success.
389 Stdout is buffered only for machine-readable modes; stderr stays live for
390 warnings and errors. Non-zero exits and exceptions replay stdout unchanged,
391 preserving existing diagnostics and exit codes. Table mode bypasses the
392 transaction entirely, including its streaming behavior.
393 """
395 def invoke(self, ctx: click.Context) -> Any:
396 try:
397 return super().invoke(ctx)
398 except click.exceptions.Exit as exc:
399 _structured_exit_code_var.set(exc.exit_code)
400 raise
402 def main(
403 self,
404 args: Sequence[str] | None = None,
405 prog_name: str | None = None,
406 complete_var: str | None = None,
407 standalone_mode: bool = True,
408 windows_expand_args: bool = True,
409 **extra: Any,
410 ) -> Any:
411 effective_args = list(sys.argv[1:] if args is None else args)
412 if _shell_completion_requested(complete_var, prog_name):
413 return super().main(
414 args=args,
415 prog_name=prog_name,
416 complete_var=complete_var,
417 standalone_mode=standalone_mode,
418 windows_expand_args=windows_expand_args,
419 **extra,
420 )
422 output_format = _requested_output_format(effective_args)
423 if output_format not in {"json", "yaml"}:
424 return super().main(
425 args=args,
426 prog_name=prog_name,
427 complete_var=complete_var,
428 standalone_mode=standalone_mode,
429 windows_expand_args=windows_expand_args,
430 **extra,
431 )
433 emissions: list[tuple[Any, str]] = []
434 emissions_token = _structured_emissions_var.set(emissions)
435 exit_code_token = _structured_exit_code_var.set(None)
436 capture = _StdoutCapture()
437 _STDOUT_CAPTURE_LOCK.acquire()
438 try:
439 try:
440 with capture:
441 result = super().main(
442 args=args,
443 prog_name=prog_name,
444 complete_var=complete_var,
445 standalone_mode=standalone_mode,
446 windows_expand_args=windows_expand_args,
447 **extra,
448 )
449 except (click.exceptions.Exit, SystemExit) as exc:
450 exit_code = getattr(exc, "exit_code", getattr(exc, "code", 1))
451 if exit_code in {None, 0}:
452 _write_structured_capture(capture.text, output_format, emissions)
453 else:
454 capture.replay()
455 raise
456 except BaseException:
457 capture.replay()
458 raise
460 captured_exit_code = _structured_exit_code_var.get()
461 if captured_exit_code not in {None, 0}:
462 capture.replay()
463 else:
464 _write_structured_capture(capture.text, output_format, emissions)
465 return result
466 finally:
467 try:
468 _structured_exit_code_var.reset(exit_code_token)
469 _structured_emissions_var.reset(emissions_token)
470 finally:
471 _STDOUT_CAPTURE_LOCK.release()
474class OutputFormatter:
475 """
476 Formats output for CLI commands.
478 Supports:
479 - Table format (human-readable)
480 - JSON format (machine-readable)
481 - YAML format (configuration-friendly)
482 """
484 def __init__(self, config: GCOConfig | None = None):
485 self.config = config or get_config()
486 self._format = self.config.output_format
488 def set_format(self, format_type: str) -> None:
489 """Set the output format."""
490 if format_type not in ("table", "json", "yaml"):
491 raise ValueError(f"Invalid format: {format_type}")
492 self._format = format_type
494 def format(self, data: Any, columns: list[str] | None = None) -> str:
495 """
496 Format data for output.
498 Args:
499 data: Data to format (dict, list, or dataclass)
500 columns: Column names for table format
502 Returns:
503 Formatted string
504 """
505 if self._format == "json":
506 return self._format_json(data)
507 if self._format == "yaml":
508 return self._format_yaml(data)
509 return self._format_table(data, columns)
511 def _format_json(self, data: Any) -> str:
512 """Format data as strict, interoperable JSON."""
513 serialized = _serialize_value(data)
514 return json.dumps(serialized, indent=2, default=str, allow_nan=False)
516 def _format_yaml(self, data: Any) -> str:
517 """Format data as YAML."""
518 serialized = _serialize_value(data)
519 return str(yaml.safe_dump(serialized, default_flow_style=False, sort_keys=False))
521 def _format_table(self, data: Any, columns: list[str] | None = None) -> str:
522 """Format data as a table."""
523 if data is None:
524 return "No data"
526 # Convert to list of dicts
527 if is_dataclass(data) and not isinstance(data, type):
528 rows = [asdict(data)]
529 elif isinstance(data, dict):
530 rows = [data]
531 elif isinstance(data, list):
532 if not data:
533 return "No results"
534 if is_dataclass(data[0]) and not isinstance(data[0], type):
535 rows = [asdict(item) for item in data]
536 elif isinstance(data[0], dict):
537 rows = data
538 else:
539 # Simple list
540 return "\n".join(str(item) for item in data)
541 else:
542 return str(data)
544 # Determine columns. `rows` is always non-empty here: every branch
545 # above either returns early (None, empty list) or assigns at least
546 # one row (a dict becomes `[data]`, even an empty `{}`).
547 if columns is None:
548 columns = list(rows[0].keys())
550 # Filter to only requested columns
551 rows = [{k: v for k, v in row.items() if k in columns} for row in rows]
553 # Calculate column widths
554 widths = {}
555 for col in columns:
556 col_values = [str(row.get(col, "")) for row in rows]
557 widths[col] = max(len(col), max(len(v) for v in col_values) if col_values else 0)
559 # Build table
560 lines = []
562 # Header
563 header = " ".join(col.upper().ljust(widths[col]) for col in columns)
564 lines.append(header)
565 lines.append("-" * len(header))
567 # Rows
568 for row in rows:
569 line = " ".join(
570 self._format_cell(row.get(col, ""), widths[col], col) for col in columns
571 )
572 lines.append(line)
574 return "\n".join(lines)
576 def _format_cell(self, value: Any, width: int, column_name: str = "") -> str:
577 """Format a single cell value."""
578 if value is None:
579 return "-".ljust(width)
580 if isinstance(value, datetime):
581 return value.strftime("%Y-%m-%d %H:%M").ljust(width)
582 if isinstance(value, bool):
583 return ("Yes" if value else "No").ljust(width)
584 if isinstance(value, float):
585 # Add dollar sign for price columns (but not stability/ratio columns)
586 col_lower = column_name.lower()
587 if "price" in col_lower and "stability" not in col_lower:
588 return f"${value:.4f}".ljust(width)
589 return f"{value:.4f}".ljust(width)
590 if isinstance(value, dict):
591 return "<dict>".ljust(width)
592 if isinstance(value, list):
593 return f"[{len(value)} items]".ljust(width)
594 return str(str(value)[:width]).ljust(width)
596 def print(self, data: Any, columns: list[str] | None = None) -> None:
597 """Format and print data, registering native machine documents."""
598 rendered = self.format(data, columns)
599 if self._format in {"json", "yaml"}:
600 emit_structured_document(
601 data,
602 output_format=self._format,
603 rendered=rendered,
604 )
605 else:
606 print(rendered)
608 def print_success(self, message: str) -> None:
609 """Print a human success message in table mode only."""
610 if self._format == "table":
611 print(f"✓ {message}")
613 def print_error(self, message: str) -> None:
614 """Print an error message."""
615 print(f"✗ {message}", file=sys.stderr)
617 def print_warning(self, message: str) -> None:
618 """Print a warning message."""
619 print(f"⚠ {message}", file=sys.stderr)
621 def print_info(self, message: str) -> None:
622 """Print a human informational message in table mode only."""
623 if self._format == "table":
624 print(f"ℹ {message}")
627# Convenience functions for common output patterns
630def format_job_table(jobs: list[Any]) -> str:
631 """Format jobs as a table."""
632 formatter = OutputFormatter()
633 return formatter.format(
634 jobs,
635 columns=[
636 "name",
637 "namespace",
638 "region",
639 "status",
640 "active_pods",
641 "succeeded_pods",
642 "failed_pods",
643 ],
644 )
647def format_capacity_table(estimates: list[Any]) -> str:
648 """Format capacity estimates as a table."""
649 formatter = OutputFormatter()
650 return formatter.format(
651 estimates,
652 columns=[
653 "instance_type",
654 "region",
655 "availability_zone",
656 "capacity_type",
657 "availability",
658 "price_per_hour",
659 "recommendation",
660 ],
661 )
664def format_file_system_table(file_systems: list[Any]) -> str:
665 """Format file systems as a table."""
666 formatter = OutputFormatter()
667 return formatter.format(
668 file_systems, columns=["file_system_id", "file_system_type", "region", "status", "dns_name"]
669 )
672def format_stack_table(stacks: list[Any]) -> str:
673 """Format regional stacks as a table."""
674 formatter = OutputFormatter()
675 return formatter.format(
676 stacks, columns=["region", "stack_name", "cluster_name", "status", "efs_file_system_id"]
677 )
680def get_output_formatter(config: GCOConfig | None = None) -> OutputFormatter:
681 """Get a configured output formatter instance."""
682 return OutputFormatter(config)