Coverage for session_buddy / utils / logging.py: 83.20%

1#!/usr/bin/env python3 

2"""Structured logging utilities for session management.""" 

3 

4from __future__ import annotations 

5 

6import json 

7import logging 

8import sys 

9import typing as t 

10from contextlib import suppress 

11from datetime import datetime 

12from pathlib import Path 

13 

14from session_buddy.di import get_sync_typed 

15from session_buddy.di.container import depends 

16 

17 

18class SessionLogger: 

19 """Structured logging for session management with context.""" 

20 

21 def __init__(self, log_dir: Path) -> None: 

22 self.log_dir = log_dir 

23 self.log_dir.mkdir(parents=True, exist_ok=True) 

24 self.log_file = ( 

25 log_dir / f"session_management_{datetime.now().strftime('%Y%m%d')}.log" 

26 ) 

27 

28 # Configure logger 

29 self.logger = logging.getLogger("session_management") 

30 self.logger.setLevel(logging.INFO) 

31 

32 formatter = logging.Formatter( 

33 "%(asctime)s | %(levelname)s | %(funcName)s:%(lineno)d | %(message)s", 

34 ) 

35 

36 # Ensure console handler exists with correct settings 

37 console_handler = _get_console_handler(self.logger) 

38 if console_handler is None: 

39 console_handler = logging.StreamHandler(sys.stderr) 

40 self.logger.addHandler(console_handler) 

41 console_handler.setLevel(logging.ERROR) 

42 console_handler.setFormatter(formatter) 

43 

44 # Ensure file handler for this log directory exists 

45 file_handler = _get_file_handler(self.logger, self.log_file) 

46 if file_handler is None: 

47 file_handler = logging.FileHandler(self.log_file) 

48 self.logger.addHandler(file_handler) 

49 file_handler.setLevel(logging.INFO) 

50 file_handler.setFormatter(formatter) 

51 

52 def info(self, message: str, **context: t.Any) -> None: 

53 """Log info with optional context.""" 

54 if context: 

55 message = f"{message} | Context: {_safe_json_serialize(context)}" 

56 self.logger.info(message) 

57 

58 def warning(self, message: str, **context: t.Any) -> None: 

59 """Log warning with optional context.""" 

60 if context: 

61 message = f"{message} | Context: {_safe_json_serialize(context)}" 

62 self.logger.warning(message) 

63 

64 def error(self, message: str, **context: t.Any) -> None: 

65 """Log error with optional context.""" 

66 if context: 

67 message = f"{message} | Context: {_safe_json_serialize(context)}" 

68 self.logger.error(message) 

69 

70 def debug(self, message: str, **context: t.Any) -> None: 

71 """Log debug with optional context.""" 

72 if context: 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true

73 message = f"{message} | Context: {_safe_json_serialize(context)}" 

74 self.logger.debug(message) 

75 

76 def exception(self, message: str, **context: t.Any) -> None: 

77 """Log exception with optional context.""" 

78 if context: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true

79 message = f"{message} | Context: {_safe_json_serialize(context)}" 

80 self.logger.error(message) 

81 

82 def critical(self, message: str, **context: t.Any) -> None: 

83 """Log critical with optional context.""" 

84 if context: 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true

85 message = f"{message} | Context: {_safe_json_serialize(context)}" 

86 self.logger.critical(message) 

87 

88 

89def get_session_logger() -> SessionLogger: 

90 """Get the global session logger instance managed by the DI container.""" 

91 with suppress(KeyError, AttributeError, RuntimeError, TypeError): 

92 # RuntimeError: when adapter requires async 

93 # TypeError: when DI has type confusion between string keys and classes 

94 logger = get_sync_typed(SessionLogger) # type: ignore[no-any-return] 

95 if isinstance(logger, SessionLogger): 95 ↛ 98line 95 didn't jump to line 98

96 return logger 

97 

98 logger = SessionLogger(_resolve_logs_dir()) 

99 depends.set(SessionLogger, logger) 

100 return logger 

101 

102 

103def _resolve_logs_dir() -> Path: 

104 # Try to get SessionPaths from DI (modern ACB pattern) 

105 with suppress(KeyError, AttributeError, RuntimeError, TypeError): 

106 from session_buddy.di.config import SessionPaths 

107 

108 paths = depends.get_sync(SessionPaths) 

109 if hasattr(paths, "logs_dir") and isinstance(paths.logs_dir, Path): 

110 paths.logs_dir.mkdir(parents=True, exist_ok=True) 

111 return paths.logs_dir 

112 

113 # Fallback: create default logs directory 

114 logs_dir = Path.home() / ".claude" / "logs" 

115 logs_dir.mkdir(parents=True, exist_ok=True) 

116 return logs_dir 

117 

118 

119def _get_console_handler( 

120 logger: logging.Logger, 

121) -> logging.StreamHandler[t.TextIO] | None: 

122 for handler in logger.handlers: 

123 if isinstance(handler, logging.StreamHandler) and not isinstance( 123 ↛ 122line 123 didn't jump to line 122 because the condition on line 123 was always true

124 handler, 

125 logging.FileHandler, 

126 ): 

127 return handler 

128 return None 

129 

130 

131def _get_file_handler( 

132 logger: logging.Logger, 

133 log_file: Path, 

134) -> logging.FileHandler | None: 

135 for handler in logger.handlers: 

136 if isinstance(handler, logging.FileHandler): 

137 try: 

138 if Path(handler.baseFilename) == log_file: 

139 return handler 

140 except Exception: 

141 continue 

142 return None 

143 

144 

145def _safe_json_serialize(obj: t.Any) -> str: 

146 """Safely serialize objects to JSON, converting non-serializable objects to strings.""" 

147 try: 

148 return json.dumps(obj) 

149 except (TypeError, ValueError): 

150 # Convert non-serializable objects to string representation 

151 if isinstance(obj, dict): 

152 return json.dumps( 

153 { 

154 # Use explicit None check for better type inference (refurb FURB168) 

155 k: str(v) 

156 if v is not None and not isinstance(v, (str, int, float, bool)) 

157 else v 

158 for k, v in obj.items() 

159 }, 

160 ) 

161 return json.dumps(str(obj))