Coverage for src / kemi / cli_writer.py: 94%

1"""CLI output writer abstraction. 

2 

3Provides three writers that share a uniform ``write``/``error`` interface: 

4 

5- :class:`ConsoleWriter` — human-readable, the default. 

6- :class:`JsonWriter` — one JSON object per line (NDJSON), for scripts. 

7- :class:`SilentWriter` — drops ``info`` messages, keeps ``error`` and 

8 ``warn``. 

9 

10The CLI entry point parses ``--json`` / ``--quiet`` flags and passes 

11the chosen writer to every subcommand as ``args.writer``. 

12""" 

13 

14from __future__ import annotations 

15 

16import json 

17import sys 

18from typing import Any, Protocol, TextIO 

19 

20 

21class Writer(Protocol): 

22 """Protocol every CLI writer satisfies.""" 

23 

24 def write(self, message: str, *, kind: str = "info", flush: bool = False, end: str = "\n") -> None: 

25 """Emit a message. ``kind`` is one of info/warn/error. 

26 

27 ``flush`` is honoured for streaming use cases (recall-stream). 

28 ``end`` controls the trailing newline (default ``"\n"``; pass 

29 ``""`` for interactive prompts). 

30 """ 

31 ... 

32 

33 def error(self, message: str) -> None: 

34 """Emit an error message. Always shown, even in --quiet mode.""" 

35 ... 

36 

37 def warn(self, message: str) -> None: 

38 """Emit a warning message.""" 

39 ... 

40 

41 

42class ConsoleWriter: 

43 """Default human-readable writer. Prints to ``stream`` (default stdout). 

44 

45 The default stream is looked up lazily so that ``capsys``/``capfd`` 

46 redirections from ``pytest`` take effect. 

47 """ 

48 

49 def __init__(self, stream: TextIO | None = None) -> None: 

50 self._explicit_stream = stream 

51 

52 def _stream(self) -> TextIO: 

53 return self._explicit_stream or sys.stdout 

54 

55 def write(self, message: str, *, kind: str = "info", flush: bool = False, end: str = "\n") -> None: 

56 s = self._stream() 

57 print(message, file=s, flush=flush) 

58 

59 def error(self, message: str) -> None: 

60 print(f"Error: {message}", file=sys.stderr) 

61 

62 def warn(self, message: str) -> None: 

63 print(f"Warning: {message}", file=sys.stderr) 

64 

65 

66class JsonWriter: 

67 """NDJSON writer — one JSON object per line, on stdout. 

68 

69 Useful for piping into ``jq`` or other tools. Use ``kind`` to 

70 distinguish info/warn/error entries. 

71 """ 

72 

73 def __init__(self, stream: TextIO | None = None) -> None: 

74 self._stream = stream or sys.stdout 

75 

76 def _emit(self, payload: dict[str, Any]) -> None: 

77 self._stream.write(json.dumps(payload, default=str) + "\n") 

78 self._stream.flush() 

79 

80 def write(self, message: str, *, kind: str = "info", flush: bool = False, end: str = "\n") -> None: 

81 self._emit({"level": kind, "message": message}) 

82 

83 def error(self, message: str) -> None: 

84 self._emit({"level": "error", "message": message}) 

85 

86 def warn(self, message: str) -> None: 

87 self._emit({"level": "warn", "message": message}) 

88 

89 

90class SilentWriter: 

91 """Drops ``info`` messages; keeps ``error`` and ``warn``.""" 

92 

93 def write(self, message: str, *, kind: str = "info", flush: bool = False, end: str = "\n") -> None: 

94 if kind != "info": 

95 print(message, file=sys.stderr) 

96 

97 def error(self, message: str) -> None: 

98 print(f"Error: {message}", file=sys.stderr) 

99 

100 def warn(self, message: str) -> None: 

101 print(f"Warning: {message}", file=sys.stderr) 

102 

103 

104def make_writer(json_mode: bool = False, quiet: bool = False) -> Writer: 

105 """Construct the writer chosen by ``--json`` and ``--quiet`` flags. 

106 

107 - ``json_mode=True, quiet=False`` → JsonWriter 

108 - ``json_mode=False, quiet=True`` → SilentWriter 

109 - ``json_mode=True, quiet=True`` → JsonWriter (quiet ignored) 

110 - default → ConsoleWriter 

111 """ 

112 if json_mode: 

113 return JsonWriter() 

114 if quiet: 

115 return SilentWriter() 

116 return ConsoleWriter()