Coverage for src / taipanstack / utils / logging.py: 100%

1""" 

2Structured logging with context. 

3 

4Provides a configured logger with support for structured output, 

5context propagation, and proper formatting. 

6""" 

7 

8import logging 

9import re 

10import sys 

11from collections.abc import Iterator, MutableMapping 

12from contextlib import AbstractContextManager, contextmanager 

13from datetime import UTC, datetime 

14from functools import lru_cache 

15from typing import Any, Literal 

16 

17from taipanstack.utils.context import get_correlation_id 

18 

19try: 

20 import structlog 

21 

22 HAS_STRUCTLOG = True 

23except ImportError: # pragma: no cover — structlog is optional 

24 HAS_STRUCTLOG = False 

25 

26 

27# Default log format 

28DEFAULT_FORMAT = "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s" 

29JSON_FORMAT = ( 

30 '{"timestamp": "%(asctime)s", "level": "%(levelname)s", ' 

31 '"logger": "%(name)s", "message": "%(message)s"}' 

32) 

33 

34SENSITIVE_KEY_PATTERNS: tuple[str, ...] = ( 

35 "password", 

36 "secret", 

37 "token", 

38 "authorization", 

39 "api_key", 

40) 

41 

42# Pre-compile the regex for O(1) checks per key instead of O(N*M) 

43_SENSITIVE_KEY_REGEX = ( 

44 re.compile("|".join(map(re.escape, SENSITIVE_KEY_PATTERNS)), re.IGNORECASE) 

45 if SENSITIVE_KEY_PATTERNS 

46 else None 

47) 

48 

49REDACTED_VALUE = "***REDACTED***" 

50 

51 

52@lru_cache(maxsize=1024) 

53def _is_sensitive(key: str, regex: re.Pattern[str] | None) -> bool: 

54 """Check if a key is sensitive using cached regex matching. 

55 

56 Args: 

57 key: The key to check. 

58 regex: The regex pattern to use for matching. 

59 

60 Returns: 

61 True if the key is sensitive, False otherwise. 

62 

63 """ 

64 if regex is None: 

65 return False 

66 return bool(regex.search(key)) 

67 

68 

69def _redact_dict(d: MutableMapping[str, Any]) -> None: 

70 """Redact sensitive keys in a dictionary in-place. 

71 

72 Args: 

73 d: The dictionary to redact. 

74 

75 """ 

76 if _SENSITIVE_KEY_REGEX is None: 

77 return 

78 

79 for key in d: 

80 if _is_sensitive(key, _SENSITIVE_KEY_REGEX): 

81 d[key] = REDACTED_VALUE 

82 

83 

84def mask_sensitive_data_processor( 

85 _logger: Any, 

86 _method: str, 

87 event_dict: MutableMapping[str, Any], 

88) -> MutableMapping[str, Any]: 

89 """Mask sensitive data in structlog event dictionaries. 

90 

91 Intercept the *event_dict* produced by structlog and replace the 

92 value of any key whose name contains a sensitive substring with 

93 ``"***REDACTED***"``. Matching is case-insensitive. 

94 

95 Args: 

96 _logger: The wrapped logger object (unused, required by structlog). 

97 _method: The name of the log method called (unused, required by structlog). 

98 event_dict: The structured event dictionary. 

99 

100 Returns: 

101 The event dictionary with sensitive values masked. 

102 

103 """ 

104 _redact_dict(event_dict) 

105 return event_dict 

106 

107 

108def correlation_id_processor( 

109 _logger: Any, 

110 _method: str, 

111 event_dict: MutableMapping[str, Any], 

112) -> MutableMapping[str, Any]: 

113 """Structlog processor to inject correlation ID into events. 

114 

115 Args: 

116 _logger: The wrapped logger object. 

117 _method: The name of the log method called. 

118 event_dict: The structured event dictionary. 

119 

120 Returns: 

121 The event dictionary with correlation_id injected if sets. 

122 

123 """ 

124 correlation_id = get_correlation_id() 

125 if correlation_id is not None: 

126 event_dict["correlation_id"] = correlation_id 

127 return event_dict 

128 

129 

130class StackLogger: 

131 """Enhanced logger with context support. 

132 

133 Provides a wrapper around standard logging with additional features 

134 like context propagation and structured output support. 

135 

136 Attributes: 

137 name: Logger name. 

138 level: Current log level. 

139 

140 """ 

141 

142 def __init__( 

143 self, 

144 name: str = "stack", 

145 level: str = "INFO", 

146 *, 

147 use_structured: bool = False, 

148 ) -> None: 

149 """Initialize the logger. 

150 

151 Args: 

152 name: Logger name. 

153 level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL). 

154 use_structured: Use structured logging if structlog is available. 

155 

156 """ 

157 self.name = name 

158 self.level = level 

159 self._context: dict[str, Any] = {} 

160 

161 if use_structured and HAS_STRUCTLOG: 

162 self._logger = structlog.get_logger(name) 

163 self._structured = True 

164 else: 

165 self._logger = logging.getLogger(name) 

166 self._logger.setLevel(getattr(logging, level.upper())) 

167 self._structured = False 

168 

169 def bind(self, **context: Any) -> "StackLogger": 

170 """Add context to logger. 

171 

172 Args: 

173 **context: Key-value pairs to add to context. 

174 

175 Returns: 

176 Self for chaining. 

177 

178 """ 

179 self._context.update(context) 

180 if self._structured and HAS_STRUCTLOG: 

181 self._logger = self._logger.bind(**context) 

182 return self 

183 

184 def unbind(self, *keys: str) -> "StackLogger": 

185 """Remove context keys. 

186 

187 Args: 

188 *keys: Keys to remove from context. 

189 

190 Returns: 

191 Self for chaining. 

192 

193 """ 

194 for key in keys: 

195 self._context.pop(key, None) 

196 if self._structured and HAS_STRUCTLOG: 

197 self._logger = self._logger.unbind(*keys) 

198 return self 

199 

200 def _format_message(self, message: str, **kwargs: Any) -> str: 

201 """Format message with context. 

202 

203 Args: 

204 message: The log message. 

205 **kwargs: Additional context for this message. 

206 

207 Returns: 

208 Formatted message string. 

209 

210 """ 

211 if not kwargs and not self._context: 

212 return message 

213 

214 context = {**self._context, **kwargs} 

215 _redact_dict(context) 

216 

217 context_str = " ".join(f"{k}={v}" for k, v in context.items()) 

218 return f"{message} | {context_str}" 

219 

220 def debug(self, message: str, **kwargs: Any) -> None: 

221 """Log a debug message. 

222 

223 Args: 

224 message: The message to log. 

225 **kwargs: Additional context. 

226 

227 """ 

228 if self._structured: 

229 self._logger.debug(message, **kwargs) 

230 else: 

231 self._logger.debug(self._format_message(message, **kwargs)) 

232 

233 def info(self, message: str, **kwargs: Any) -> None: 

234 """Log an info message. 

235 

236 Args: 

237 message: The message to log. 

238 **kwargs: Additional context. 

239 

240 """ 

241 if self._structured: 

242 self._logger.info(message, **kwargs) 

243 else: 

244 self._logger.info(self._format_message(message, **kwargs)) 

245 

246 def warning(self, message: str, **kwargs: Any) -> None: 

247 """Log a warning message. 

248 

249 Args: 

250 message: The message to log. 

251 **kwargs: Additional context. 

252 

253 """ 

254 if self._structured: 

255 self._logger.warning(message, **kwargs) 

256 else: 

257 self._logger.warning(self._format_message(message, **kwargs)) 

258 

259 def error(self, message: str, **kwargs: Any) -> None: 

260 """Log an error message. 

261 

262 Args: 

263 message: The message to log. 

264 **kwargs: Additional context. 

265 

266 """ 

267 if self._structured: 

268 self._logger.error(message, **kwargs) 

269 else: 

270 self._logger.error(self._format_message(message, **kwargs)) 

271 

272 def critical(self, message: str, **kwargs: Any) -> None: 

273 """Log a critical message. 

274 

275 Args: 

276 message: The message to log. 

277 **kwargs: Additional context. 

278 

279 """ 

280 if self._structured: 

281 self._logger.critical(message, **kwargs) 

282 else: 

283 self._logger.critical(self._format_message(message, **kwargs)) 

284 

285 def exception(self, message: str, **kwargs: Any) -> None: 

286 """Log an exception with traceback. 

287 

288 Args: 

289 message: The message to log. 

290 **kwargs: Additional context. 

291 

292 """ 

293 if self._structured: 

294 self._logger.exception(message, **kwargs) 

295 else: 

296 self._logger.exception(self._format_message(message, **kwargs)) 

297 

298 

299def setup_logging( 

300 level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO", 

301 *, 

302 format_type: Literal["simple", "detailed", "json"] = "detailed", 

303 log_file: str | None = None, 

304 use_structured: bool = False, 

305) -> None: 

306 """Configure the root logger. 

307 

308 Args: 

309 level: Log level to set. 

310 format_type: Output format type. 

311 log_file: Optional file to log to. 

312 use_structured: Use structlog if available. 

313 

314 """ 

315 # Configure structlog if available and requested 

316 if use_structured and HAS_STRUCTLOG: 

317 structlog.configure( 

318 processors=[ 

319 structlog.stdlib.filter_by_level, 

320 structlog.stdlib.add_logger_name, 

321 structlog.stdlib.add_log_level, 

322 structlog.processors.TimeStamper(fmt="iso"), 

323 correlation_id_processor, 

324 mask_sensitive_data_processor, 

325 structlog.processors.StackInfoRenderer(), 

326 structlog.processors.format_exc_info, 

327 structlog.processors.UnicodeDecoder(), 

328 structlog.processors.JSONRenderer(), 

329 ], 

330 wrapper_class=structlog.stdlib.BoundLogger, 

331 context_class=dict, 

332 logger_factory=structlog.stdlib.LoggerFactory(), 

333 ) 

334 return 

335 

336 # Standard logging configuration 

337 match format_type: 

338 case "simple": 

339 log_format = "%(levelname)s: %(message)s" 

340 case "json": 

341 log_format = JSON_FORMAT 

342 case _: 

343 log_format = DEFAULT_FORMAT 

344 

345 handlers: list[logging.Handler] = [logging.StreamHandler(sys.stdout)] 

346 

347 if log_file: 

348 handlers.append(logging.FileHandler(log_file, encoding="utf-8")) 

349 

350 logging.basicConfig( 

351 level=getattr(logging, level.upper()), 

352 format=log_format, 

353 handlers=handlers, 

354 force=True, 

355 ) 

356 

357 

358def get_logger( 

359 name: str = "stack", 

360 *, 

361 level: str = "INFO", 

362 use_structured: bool = False, 

363) -> StackLogger: 

364 """Get a configured logger instance. 

365 

366 Args: 

367 name: Logger name. 

368 level: Log level. 

369 use_structured: Use structlog if available. 

370 

371 Returns: 

372 Configured StackLogger instance. 

373 

374 Example: 

375 >>> logger = get_logger("my_module") 

376 >>> logger.bind(request_id="123").info("Processing request") 

377 

378 """ 

379 return StackLogger(name, level, use_structured=use_structured) 

380 

381 

382def log_operation( 

383 operation: str, 

384 *, 

385 logger: StackLogger | None = None, 

386 level: str = "INFO", 

387 expected_exceptions: tuple[type[Exception], ...] | type[Exception] = Exception, 

388) -> AbstractContextManager[StackLogger]: 

389 """Context manager for logging operations. 

390 

391 Args: 

392 operation: Name of the operation. 

393 logger: Logger to use (creates one if not provided). 

394 level: Log level for messages. 

395 expected_exceptions: Exceptions to catch and log as failures. 

396 

397 Yields: 

398 The logger instance. 

399 

400 Example: 

401 >>> with log_operation("setup") as logger: 

402 ... logger.info("Setting up environment") 

403 

404 """ 

405 

406 @contextmanager 

407 def _log_context() -> Iterator[StackLogger]: 

408 nonlocal logger 

409 if logger is None: 

410 logger = get_logger() 

411 

412 start_time = datetime.now(UTC) 

413 logger.bind(operation=operation) 

414 

415 log_method = getattr(logger, level.lower()) 

416 log_method(f"Starting: {operation}") 

417 

418 try: 

419 yield logger 

420 duration = (datetime.now(UTC) - start_time).total_seconds() 

421 log_method(f"Completed: {operation}", duration_seconds=duration) 

422 except expected_exceptions as e: 

423 duration = (datetime.now(UTC) - start_time).total_seconds() 

424 logger.exception( 

425 f"Failed: {operation}", 

426 duration_seconds=duration, 

427 error=str(e), 

428 ) 

429 raise 

430 finally: 

431 logger.unbind("operation") 

432 

433 return _log_context()