Coverage for  / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / infra / emitter.py: 91%

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Output emitter adapters.""" 

5 

6from __future__ import annotations 

7 

8import sys 

9from typing import Any 

10 

11import structlog 

12 

13from bijux_cli.core.enums import LogLevel, OutputFormat 

14from bijux_cli.infra.serializer import serializer_for 

15 

16 

17class ConsoleEmitter: 

18 """Emitter that serializes and writes payloads to stdout.""" 

19 

20 def __init__( 

21 self, 

22 telemetry: Any, 

23 output_format: OutputFormat, 

24 ) -> None: 

25 """Initialize the console emitter.""" 

26 self._telemetry = telemetry 

27 self._output_format = output_format 

28 self._logger = structlog.get_logger(__name__) 

29 

30 def emit( 

31 self, 

32 payload: Any, 

33 *, 

34 fmt: OutputFormat, 

35 pretty: bool = False, 

36 level: LogLevel = LogLevel.INFO, 

37 message: str = "Emitting output", 

38 output: str | None = None, 

39 emit_output: bool = True, 

40 emit_diagnostics: bool = False, 

41 **context: Any, 

42 ) -> None: 

43 """Serialize and emit a payload.""" 

44 if not emit_output: 

45 return 

46 

47 output_format = fmt 

48 serializer = serializer_for(output_format, self._telemetry) 

49 try: 

50 output_str = serializer.dumps(payload, fmt=output_format, pretty=pretty) 

51 except Exception as error: 

52 self._logger.error("Serialization failed", error=str(error), **context) 

53 raise RuntimeError(f"Serialization failed: {error}") from error 

54 

55 stripped = output_str.rstrip("\n") 

56 if output: 

57 with open(output, "w", encoding="utf-8") as f: 

58 f.write(stripped) 

59 else: 

60 print(stripped, file=sys.stdout, flush=True) 

61 

62 try: 

63 format_name = output_format.value 

64 self._telemetry.event( 

65 "output_emitted", 

66 {"format": format_name, "size_chars": len(stripped)}, 

67 ) 

68 except Exception as tel_err: 

69 self._logger.debug("Telemetry failed", error=str(tel_err), **context) 

70 

71 def flush(self) -> None: 

72 """Flushes standard output.""" 

73 sys.stdout.flush() 

74 

75 

76class NullEmitter: 

77 """Emitter that discards output.""" 

78 

79 def emit( 

80 self, 

81 payload: Any, 

82 *, 

83 fmt: OutputFormat, 

84 pretty: bool = False, 

85 level: LogLevel = LogLevel.INFO, 

86 message: str = "Emitting output", 

87 output: str | None = None, 

88 emit_output: bool = True, 

89 emit_diagnostics: bool = False, 

90 **context: Any, 

91 ) -> None: 

92 """Drop emitted payloads.""" 

93 _ = ( 

94 payload, 

95 fmt, 

96 pretty, 

97 level, 

98 message, 

99 output, 

100 emit_output, 

101 emit_diagnostics, 

102 ) 

103 _ = context 

104 return None 

105 

106 def flush(self) -> None: 

107 """No-op flush for the null emitter.""" 

108 return None 

109 

110 

111__all__ = ["ConsoleEmitter", "NullEmitter"]