Coverage for src / template_python / logging.py: 0%

1""" 

2Loguru-based logging configuration and environment settings for template_python. 

3 

4This module provides a unified interface for configuring application-level logging 

5using loguru and Pydantic settings. It handles dynamic OpenTelemetry formatting, 

6across the codebase and build environments. 

7""" 

8 

9from __future__ import annotations 

10 

11import contextlib 

12import json 

13import sys 

14import traceback 

15from collections.abc import Callable 

16from typing import Any, ClassVar, Literal 

17 

18from loguru import logger 

19from pydantic import Field, field_validator 

20from pydantic_settings import BaseSettings, SettingsConfigDict 

21 

22from template_python.compat import opentelemetry_trace 

23 

24__all__ = ["LoggingSettings", "configure_logger", "logger"] 

25 

26_HANDLER_ID: int | None = None 

27 

28 

29class LoggingSettings(BaseSettings): 

30 """ 

31 Settings model for configuring the loguru logging infrastructure. 

32 

33 This Pydantic model loads configuration from environment variables prefixed 

34 with ``TEMPLATE_PYTHON__LOGGING__`` and provides typed fields for controlling 

35 log output, formatting, and OpenTelemetry integration. 

36 

37 Example: 

38 .. code-block:: python 

39 

40 from template_python.logging import LoggingSettings, configure_logger 

41 

42 settings = LoggingSettings(enabled=True, level="DEBUG") 

43 configure_logger(settings) 

44 """ 

45 

46 enabled: bool = Field( 

47 default=False, 

48 description=( 

49 "Whether to enable template_python loguru logging across the application." 

50 ), 

51 ) 

52 clear_loggers: bool = Field( 

53 default=False, 

54 description=( 

55 "If true, removes all existing loguru handlers before configuring new ones." 

56 ), 

57 ) 

58 sink: str | Any = Field( 

59 default=sys.stdout, 

60 description=( 

61 "The output sink for log messages. Can be an object or string " 

62 "alias ('stdout')." 

63 ), 

64 ) 

65 level: str = Field( 

66 default="INFO", 

67 description="The minimum severity level for emitted log messages.", 

68 ) 

69 otel_formatting: Literal["auto", "enable", "disable"] = Field( 

70 default="auto", 

71 description=( 

72 "Controls OpenTelemetry JSON formatting. 'auto' enables it if " 

73 "otel is installed." 

74 ), 

75 ) 

76 format: str | Callable[..., Any] | None = Field( 

77 default="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | " 

78 "<cyan>{name}</cyan>:<cyan>{function}</cyan> - <level>{message}</level>\n", 

79 description=( 

80 "The log format string or function to use when otel formatting is disabled." 

81 ), 

82 ) 

83 filter: Any = Field( 

84 default=True, 

85 description=( 

86 "Filters log records. Defaults to True to filter by the " 

87 "'template_python' prefix." 

88 ), 

89 ) 

90 enqueue: bool = Field( 

91 default=True, 

92 description="Whether to enable thread-safe asynchronous logging.", 

93 ) 

94 kwargs: dict[str, Any] = Field( 

95 default_factory=dict, 

96 description=( 

97 "Additional keyword arguments to pass directly to loguru's add() method." 

98 ), 

99 ) 

100 

101 model_config: ClassVar[SettingsConfigDict] = SettingsConfigDict( 

102 env_prefix="TEMPLATE_PYTHON__LOGGING__", 

103 env_nested_delimiter="__", 

104 ) 

105 """Pydantic configuration dict dictating environment variable prefixes.""" 

106 

107 @field_validator("sink", mode="before") 

108 @classmethod 

109 def _parse_sink(cls, value: Any) -> Any: 

110 # Convert string aliases for stdout/stderr to actual objects. 

111 if isinstance(value, str): 

112 mapping = { 

113 "stdout": sys.stdout, 

114 "sys.stdout": sys.stdout, 

115 "stderr": sys.stderr, 

116 "sys.stderr": sys.stderr, 

117 } 

118 return mapping.get(value.lower(), value) 

119 return value 

120 

121 

122def _otel_formatter(record: dict[str, Any]) -> str: 

123 # Format the log record as an OpenTelemetry compliant JSON string. 

124 trace_id = span_id = trace_flags = None 

125 

126 if opentelemetry_trace: 

127 span = opentelemetry_trace.get_current_span() 

128 context = span.get_span_context() 

129 if context.is_valid: 

130 trace_id = format(context.trace_id, "032x") 

131 span_id = format(context.span_id, "016x") 

132 trace_flags = format(context.trace_flags, "02x") 

133 

134 log_record = { 

135 "timestamp": record["time"].isoformat(), 

136 "severity_text": record["level"].name, 

137 "body": record["message"], 

138 "resource": {"service.name": "template_python"}, 

139 "attributes": { 

140 "module": record["name"], 

141 "function": record["function"], 

142 "line": record["line"], 

143 **record["extra"], 

144 }, 

145 } 

146 

147 if record.get("exception"): 

148 exception = record["exception"] 

149 log_record["attributes"]["exception.type"] = exception.type.__name__ 

150 log_record["attributes"]["exception.message"] = str(exception.value) 

151 log_record["attributes"]["exception.stacktrace"] = "".join( 

152 traceback.format_exception( 

153 exception.type, exception.value, exception.traceback 

154 ) 

155 ) 

156 

157 if trace_id: 

158 log_record.update( 

159 { 

160 "trace_id": trace_id, 

161 "span_id": span_id, 

162 "trace_flags": trace_flags, 

163 } 

164 ) 

165 

166 # Escape braces so loguru doesn't interpret the JSON string as a format string 

167 return json.dumps(log_record).replace("{", "{{").replace("}", "}}") + "\n" 

168 

169 

170def configure_logger(settings: LoggingSettings | None = None) -> None: 

171 """ 

172 Initializes the loguru logger with the provided settings or from the environment. 

173 

174 This function configures the global loguru logger instance based on the provided 

175 ``LoggingSettings``. It handles enabling/disabling the logger, managing sinks, 

176 and injecting the appropriate formatter (including OpenTelemetry). 

177 

178 Example: 

179 .. code-block:: python 

180 

181 from template_python.logging import configure_logger, LoggingSettings 

182 

183 configure_logger(LoggingSettings(level="DEBUG")) 

184 

185 :param settings: An optional instance of ``LoggingSettings``. If not provided, 

186 settings are automatically loaded from the environment. 

187 :return: None 

188 :raises ImportError: If OpenTelemetry formatting is explicitly enabled but the 

189 package is not installed. 

190 """ 

191 global _HANDLER_ID # noqa: PLW0603 

192 

193 settings = settings or LoggingSettings() 

194 

195 if not settings.enabled: 

196 logger.disable("template_python") 

197 return 

198 

199 logger.enable("template_python") 

200 

201 if settings.clear_loggers: 

202 logger.remove() 

203 _HANDLER_ID = None 

204 elif isinstance(_HANDLER_ID, int): 

205 with contextlib.suppress(ValueError): 

206 logger.remove(_HANDLER_ID) 

207 _HANDLER_ID = None 

208 

209 use_otel = settings.otel_formatting == "enable" or ( 

210 settings.otel_formatting == "auto" and opentelemetry_trace is not None 

211 ) 

212 if settings.otel_formatting == "enable" and opentelemetry_trace is None: 

213 raise ImportError( 

214 "OpenTelemetry is not installed but 'otel_formatting' was set to 'enable'." 

215 ) 

216 

217 log_format = _otel_formatter if use_otel else settings.format 

218 filter_val = "template_python" if settings.filter is True else settings.filter 

219 

220 if isinstance(filter_val, (list, tuple)): 

221 prefixes = tuple(filter_val) 

222 

223 def final_filter(record: dict[str, Any]) -> bool: 

224 return bool(record["name"] and record["name"].startswith(prefixes)) 

225 

226 elif isinstance(filter_val, str): 

227 

228 def final_filter(record: dict[str, Any]) -> bool: 

229 return bool(record["name"] and record["name"].startswith(filter_val)) 

230 

231 else: 

232 final_filter = None if filter_val is False else filter_val # type: ignore[assignment] 

233 

234 _HANDLER_ID = logger.add( 

235 settings.sink, # type: ignore[arg-type] 

236 level=settings.level, 

237 filter=final_filter, # type: ignore[arg-type] 

238 format=log_format, # type: ignore[arg-type] 

239 enqueue=settings.enqueue, 

240 **settings.kwargs, 

241 )