Coverage for  / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / services / logging / observability.py: 100%

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Provides the concrete implementation of the observability and logging service. 

5 

6This module defines the `Observability` class, which implements the 

7`ObservabilityProtocol`. It serves as the primary interface for structured 

8logging throughout the application, using `structlog` as its underlying engine. 

9It can also be configured to forward log entries to a telemetry backend, 

10unifying logging and event tracking. 

11""" 

12 

13from __future__ import annotations 

14 

15from typing import Any, Self 

16 

17from injector import inject 

18import structlog 

19from structlog.typing import FilteringBoundLogger 

20 

21from bijux_cli.core.enums import LogLevel 

22from bijux_cli.services.contracts import ObservabilityProtocol, TelemetryProtocol 

23from bijux_cli.services.errors import ServiceError 

24 

25 

26class Observability(ObservabilityProtocol): 

27 """A structured logging service integrating `structlog` and telemetry. 

28 

29 This class wraps a `structlog` logger to produce structured log entries. 

30 If configured with a telemetry backend, it also forwards these events for 

31 analytics and monitoring. 

32 

33 Attributes: 

34 _logger (FilteringBoundLogger): The underlying `structlog` logger instance. 

35 _telemetry (TelemetryProtocol): The telemetry service for event forwarding. 

36 """ 

37 

38 @inject 

39 def __init__( 

40 self, *, log_level: LogLevel = LogLevel.INFO, telemetry: TelemetryProtocol 

41 ) -> None: 

42 """Initializes the observability service. 

43 

44 Args: 

45 log_level (str): The log level used for this instance. 

46 telemetry (TelemetryProtocol): The telemetry sink used for events. 

47 """ 

48 self._logger: FilteringBoundLogger = structlog.get_logger("bijux_cli") 

49 self._telemetry = telemetry 

50 _ = log_level 

51 

52 def set_telemetry(self, telemetry: TelemetryProtocol) -> Self: 

53 """Attaches a telemetry backend for forwarding log events. 

54 

55 This allows the service to be "upgraded" from a simple logger to a full 

56 observability tool after its initial creation. 

57 

58 Args: 

59 telemetry (TelemetryProtocol): The telemetry service to receive events. 

60 

61 Returns: 

62 Self: The service instance, allowing for method chaining. 

63 """ 

64 self._telemetry = telemetry 

65 return self 

66 

67 @classmethod 

68 def setup( 

69 cls, *, log_level: LogLevel = LogLevel.INFO, telemetry: TelemetryProtocol 

70 ) -> Self: 

71 """Instantiates and configures an `Observability` service. 

72 

73 Args: 

74 log_level (str): The configured log level. 

75 telemetry (TelemetryProtocol): The telemetry sink for events. 

76 

77 Returns: 

78 Self: A new, configured `Observability` instance. 

79 """ 

80 return cls(log_level=log_level, telemetry=telemetry) 

81 

82 def get_logger(self) -> FilteringBoundLogger: 

83 """Retrieves the underlying `structlog` logger instance. 

84 

85 Returns: 

86 FilteringBoundLogger: The `structlog` logger, which can be used 

87 directly if needed. 

88 """ 

89 return self._logger 

90 

91 def bind(self, **kwargs: Any) -> Self: 

92 """Binds context key-value pairs to all subsequent log entries. 

93 

94 Args: 

95 **kwargs (Any): Context values to include in each log entry. 

96 

97 Returns: 

98 Self: The service instance, allowing for method chaining. 

99 """ 

100 self._logger = self._logger.bind(**kwargs) 

101 return self 

102 

103 def log( 

104 self, 

105 level: str, 

106 msg: str, 

107 *, 

108 extra: dict[str, Any] | None = None, 

109 ) -> Self: 

110 """Logs a structured message and emits a corresponding telemetry event. 

111 

112 Args: 

113 level (str): The severity level of the log (e.g., 'debug', 'info', 

114 'warning', 'error', 'critical'). 

115 msg (str): The log message. 

116 extra (dict[str, Any] | None): Additional context to include in the 

117 log entry. 

118 

119 Returns: 

120 Self: The service instance, allowing for method chaining. 

121 

122 Raises: 

123 ServiceError: If `level` is not a valid log level name. 

124 """ 

125 log_func = getattr(self._logger, level.lower(), None) 

126 if not callable(log_func): 

127 raise ServiceError(f"Invalid log level: {level}") 

128 

129 log_context = extra or {} 

130 if log_context: 

131 log_func(msg, **log_context) 

132 else: 

133 log_func(msg) 

134 

135 telemetry_payload = {"level": level, "message": msg} 

136 telemetry_payload.update(log_context) 

137 self._telemetry.event("LOG_EMITTED", telemetry_payload) 

138 

139 return self 

140 

141 def close(self) -> None: 

142 """Logs the shutdown of the observability service. 

143 

144 Note: 

145 In this implementation, this method only logs a debug message and 

146 does not perform resource cleanup like flushing. Flushing is 

147 handled by the telemetry service's own lifecycle methods. 

148 """ 

149 self._logger.debug("Observability shutdown") 

150 

151 

152__all__ = ["Observability"]