Source code for kelvin.sdk.output

"""Output service for controlling CLI output behavior.

This module provides the Output class that handles:
- JSON mode vs interactive mode
- Progress indicators (static status messages)
- User confirmations
- Consistent error output to stderr
"""

import json
from collections.abc import Generator
from contextlib import contextmanager
from dataclasses import dataclass, field
from typing import Any, Optional

import rich_click as click
from rich.console import Console


[docs] @dataclass class Output: """Controls output behavior based on mode. Attributes: json_mode: If True, suppress interactive output and format results as JSON. assume_yes: If True, skip confirmation prompts (used with --yes flag). verbose: If True, verbose logging is enabled. """ json_mode: bool = False assume_yes: bool = False verbose: bool = False _console: Console = field(default_factory=Console, repr=False) _stderr_console: Console = field(default_factory=lambda: Console(stderr=True), repr=False) def __post_init__(self) -> None: # Recreate console with quiet mode if json_mode if self.json_mode: self._console = Console(quiet=True) @property def console(self) -> Console: """Access the console for advanced formatting (tables, trees, etc.).""" return self._console
[docs] @contextmanager def status(self, message: str) -> Generator[None]: """Print a static status message to stderr, nothing in JSON mode. This was originally a Rich spinner (Console.status), but structlog messages and subprocess output (e.g. docker buildx) corrupted the spinner rendering. Changing to a static message for now. Args: message: Status message to display. Example: with out.status("Loading..."): do_something_slow() """ if self.json_mode: yield else: self._stderr_console.print(f"[dim]{message}[/dim]") yield
[docs] def print(self, *args: Any, **kwargs: Any) -> None: """Print to console (suppressed in JSON mode). Use for interactive output like tables, formatted results, etc. """ if not self.json_mode: self._console.print(*args, **kwargs)
[docs] def print_json(self, data: Any) -> None: """Print JSON output to stdout. Always prints regardless of mode. Use for command results in --json mode. Args: data: Data to serialize as JSON. Can be a dict, list, or dataclass. """ # Handle dataclasses if hasattr(data, "__dataclass_fields__"): from dataclasses import asdict data = asdict(data) print(json.dumps(data, indent=2, default=str))
[docs] def error(self, message: str) -> None: """Print error to stderr (always visible). Args: message: Error message to display. """ self._stderr_console.print(f"[red]Error:[/red] {message}")
[docs] def warning(self, message: str) -> None: """Print warning to stderr (always visible). Args: message: Warning message to display. """ self._stderr_console.print(f"[yellow]Warning:[/yellow] {message}")
[docs] def warning_detail(self, message: str) -> None: """Print indented warning detail to stderr (bold). Args: message: Detail line to display below a warning. """ self._stderr_console.print(f" [bold]{message}[/]")
[docs] def debug(self, message: str) -> None: """Print debug info to stderr. Only shown when verbose mode is enabled. Args: message: Debug message to display. """ if not self.verbose: return self._stderr_console.print(f"[dim]{message}[/dim]")
[docs] def confirm(self, message: str, default: bool = False) -> bool: """Prompt for confirmation. Behavior: - Interactive mode: prompts user - ``--yes``: returns True - JSON mode without ``--yes``: returns False (no output — caller decides) - Non-interactive without ``--yes``: returns False with stderr hint Args: message: Confirmation prompt message. default: Default value if user just presses Enter. Returns: True if confirmed, False otherwise. """ import sys if self.assume_yes: return True if self.json_mode: return False # Check if stdin is interactive if not sys.stdin.isatty(): self._stderr_console.print( "[dim]Non-interactive session detected. Use --yes (-y) to skip confirmation.[/dim]" ) return False return click.confirm(message, default=default)
[docs] def abort_if_not_confirmed(self, message: str) -> None: """Prompt for confirmation, abort if not confirmed. Like click.confirm(..., abort=True) but JSON-aware. Raises SystemExit(0) if not confirmed. Args: message: Confirmation prompt message. Raises: SystemExit: With code 0 if user cancels. """ if not self.confirm(message): if self.json_mode: self.print_json( { "status": "cancelled", "message": message, "hint": "Use --yes (-y) to skip confirmation", } ) else: self._stderr_console.print("[yellow]Cancelled[/yellow]") raise SystemExit(0)
[docs] def success(self, message: str) -> None: """Print success message to stderr (suppressed in JSON mode). Args: message: Success message to display. """ if not self.json_mode: self._stderr_console.print(f"[green]✓[/green] {message}")
[docs] def prompt(self, message: str, default: Optional[str] = None, hide_input: bool = False) -> str: """Prompt user for input. In JSON mode, uses default or raises error if no default provided. Args: message: Prompt message. default: Default value. hide_input: If True, hide input (for passwords). Returns: User input string. Raises: SystemExit: If in JSON mode with no default. """ if self.json_mode: if default is not None: return default self.error("Input required but running in non-interactive mode") self.print_json({"status": "error", "reason": "input_required"}) raise SystemExit(1) result: str = click.prompt(message, default=default, hide_input=hide_input) return result