Coverage for src/usaspending/logging_config.py: 57%

1"""Logging configuration for USASpending API client.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import sys 

7from typing import Dict, Any, Optional 

8 

9 

10class USASpendingLogger: 

11 """Centralized logging configuration for USASpending API client.""" 

12 

13 _loggers: Dict[str, logging.Logger] = {} 

14 _configured = False 

15 

16 @classmethod 

17 def configure( 

18 self, 

19 level: str = "INFO", 

20 debug_mode: bool = False, 

21 log_format: Optional[str] = None, 

22 log_file: Optional[str] = None, 

23 ) -> None: 

24 """ 

25 Configure logging for the USASpending API client. 

26 

27 Args: 

28 level: Base logging level (DEBUG, INFO, WARNING, ERROR) 

29 debug_mode: Enable verbose debug logging for API calls and internals 

30 log_format: Custom log format string 

31 log_file: Optional file to write logs to (in addition to console) 

32 """ 

33 if self._configured: 

34 return 

35 

36 # Set root logger level 

37 root_logger = logging.getLogger("usaspending") 

38 

39 # Set level based on debug mode 

40 if debug_mode: 

41 root_logger.setLevel(logging.DEBUG) 

42 level = "DEBUG" 

43 else: 

44 root_logger.setLevel(getattr(logging, level.upper(), logging.INFO)) 

45 

46 # Remove existing handlers 

47 for handler in root_logger.handlers[:]: 

48 root_logger.removeHandler(handler) 

49 

50 # Create formatter 

51 if not log_format: 

52 if debug_mode: 

53 log_format = ( 

54 "%(asctime)s - %(name)s - %(levelname)s - " 

55 "%(filename)s:%(lineno)d - %(funcName)s() - %(message)s" 

56 ) 

57 else: 

58 log_format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" 

59 

60 formatter = logging.Formatter(log_format) 

61 

62 # Console handler 

63 console_handler = logging.StreamHandler(sys.stdout) 

64 console_handler.setFormatter(formatter) 

65 root_logger.addHandler(console_handler) 

66 

67 # File handler if specified 

68 if log_file: 

69 file_handler = logging.FileHandler(log_file) 

70 file_handler.setFormatter(formatter) 

71 root_logger.addHandler(file_handler) 

72 

73 # Configure specific loggers for different components 

74 component_levels = { 

75 "usaspending.client": logging.DEBUG if debug_mode else logging.INFO, 

76 "usaspending.queries": logging.DEBUG if debug_mode else logging.INFO, 

77 "usaspending.models": logging.DEBUG if debug_mode else logging.WARNING, 

78 "usaspending.utils.rate_limit": logging.DEBUG 

79 if debug_mode 

80 else logging.WARNING, 

81 "usaspending.utils.retry": logging.DEBUG if debug_mode else logging.WARNING, 

82 "usaspending.cache": logging.DEBUG if debug_mode else logging.WARNING, 

83 } 

84 

85 for logger_name, logger_level in component_levels.items(): 

86 logger = logging.getLogger(logger_name) 

87 logger.setLevel(logger_level) 

88 # Prevent duplicate messages 

89 logger.propagate = True 

90 

91 # Mark as configured 

92 self._configured = True 

93 

94 # Log configuration 

95 config_logger = self.get_logger("usaspending.config") 

96 config_logger.info(f"Logging configured - Level: {level}, Debug: {debug_mode}") 

97 if log_file: 

98 config_logger.info(f"Logging to file: {log_file}") 

99 

100 @classmethod 

101 def get_logger(cls, name: str) -> logging.Logger: 

102 """ 

103 Get a logger instance for the given name. 

104 

105 Args: 

106 name: Logger name (typically module name) 

107 

108 Returns: 

109 Configured logger instance 

110 """ 

111 if name not in cls._loggers: 

112 cls._loggers[name] = logging.getLogger(name) 

113 return cls._loggers[name] 

114 

115 @classmethod 

116 def is_debug_enabled(cls) -> bool: 

117 """Check if debug logging is enabled.""" 

118 root_logger = logging.getLogger("usaspending") 

119 return root_logger.isEnabledFor(logging.DEBUG) 

120 

121 @classmethod 

122 def reset(cls) -> None: 

123 """Reset logging configuration (mainly for testing).""" 

124 cls._configured = False 

125 cls._loggers.clear() 

126 

127 # Clear all handlers from our loggers 

128 for logger_name in [ 

129 "usaspending", 

130 "usaspending.client", 

131 "usaspending.queries", 

132 "usaspending.models", 

133 "usaspending.utils", 

134 "usaspending.cache", 

135 ]: 

136 logger = logging.getLogger(logger_name) 

137 for handler in logger.handlers[:]: 

138 logger.removeHandler(handler) 

139 

140 

141def get_logger(name: str) -> logging.Logger: 

142 """ 

143 Convenience function to get a logger. 

144 

145 Args: 

146 name: Logger name 

147 

148 Returns: 

149 Logger instance 

150 """ 

151 return USASpendingLogger.get_logger(name) 

152 

153 

154def log_api_request( 

155 logger: logging.Logger, 

156 method: str, 

157 url: str, 

158 params: Optional[Dict[str, Any]] = None, 

159 json_data: Optional[Dict[str, Any]] = None, 

160) -> None: 

161 """ 

162 Log an API request with appropriate detail level. 

163 

164 Args: 

165 logger: Logger instance 

166 method: HTTP method 

167 url: Request URL 

168 params: Query parameters 

169 json_data: JSON payload 

170 """ 

171 if logger.isEnabledFor(logging.DEBUG): 

172 logger.debug(f"API Request: {method} {url}") 

173 if params: 

174 logger.debug(f"Query params: {params}") 

175 if json_data: 

176 logger.debug(f"JSON payload: {json_data}") 

177 else: 

178 logger.info(f"API Request: {method} {url}") 

179 

180 

181def log_api_response( 

182 logger: logging.Logger, 

183 status_code: int, 

184 response_size: Optional[int] = None, 

185 duration: Optional[float] = None, 

186 error: Optional[str] = None, 

187) -> None: 

188 """ 

189 Log an API response with appropriate detail level. 

190 

191 Args: 

192 logger: Logger instance 

193 status_code: HTTP status code 

194 response_size: Response size in bytes 

195 duration: Request duration in seconds 

196 error: Error message if applicable 

197 """ 

198 if error: 

199 logger.error(f"API Response: {status_code} - Error: {error}") 

200 elif status_code >= 400: 

201 logger.warning(f"API Response: {status_code}") 

202 else: 

203 msg_parts = [f"API Response: {status_code}"] 

204 if duration is not None: 

205 msg_parts.append(f"({duration:.3f}s)") 

206 if response_size is not None and logger.isEnabledFor(logging.DEBUG): 

207 msg_parts.append(f"- {response_size} bytes") 

208 

209 if status_code >= 300: 

210 logger.warning(" ".join(msg_parts)) 

211 else: 

212 logger.info(" ".join(msg_parts)) 

213 

214 

215def log_query_execution( 

216 logger: logging.Logger, 

217 query_type: str, 

218 filters_count: int, 

219 endpoint: str, 

220 page: int = 1, 

221) -> None: 

222 """ 

223 Log query execution details. 

224 

225 Args: 

226 logger: Logger instance 

227 query_type: Type of query (e.g., "AwardsSearch") 

228 filters_count: Number of filters applied 

229 endpoint: API endpoint 

230 page: Page number 

231 """ 

232 if logger.isEnabledFor(logging.DEBUG): 

233 logger.debug( 

234 f"Executing {query_type} query - {filters_count} filters, " 

235 f"endpoint: {endpoint}, page: {page}" 

236 ) 

237 else: 

238 logger.info(f"Executing {query_type} query - {filters_count} filters")