Coverage for packages / core / common / logging.py: 61%

1"""Structured logging setup for TipSharks Elo API. 

2 

3Provides JSON and text logging with contextual fields for traceability. 

4""" 

5 

6import logging 

7import sys 

8from typing import Any 

9 

10from pythonjsonlogger import jsonlogger 

11 

12from packages.core.common.settings import get_settings 

13 

14 

15class CustomJsonFormatter(jsonlogger.JsonFormatter): 

16 """Custom JSON formatter with additional fields.""" 

17 

18 def add_fields( 

19 self, 

20 log_record: dict[str, Any], 

21 record: logging.LogRecord, 

22 message_dict: dict[str, Any], 

23 ) -> None: 

24 """Add custom fields to log record.""" 

25 super().add_fields(log_record, record, message_dict) 

26 log_record["level"] = record.levelname 

27 log_record["logger"] = record.name 

28 

29 

30def setup_logging() -> None: 

31 """Configure application logging based on settings.""" 

32 settings = get_settings() 

33 

34 # Get root logger 

35 root_logger = logging.getLogger() 

36 root_logger.setLevel(getattr(logging, settings.logging.level.upper())) 

37 

38 # Remove existing handlers 

39 root_logger.handlers = [] 

40 

41 # Create console handler 

42 handler = logging.StreamHandler(sys.stdout) 

43 

44 # Set formatter based on config 

45 if settings.logging.format.lower() == "json": 

46 formatter = CustomJsonFormatter( 

47 "%(timestamp)s %(level)s %(name)s %(message)s", 

48 datefmt="%Y-%m-%d %H:%M:%S", 

49 ) 

50 else: 

51 formatter = logging.Formatter( 

52 "%(asctime)s - %(name)s - %(levelname)s - %(message)s", 

53 datefmt="%Y-%m-%d %H:%M:%S", 

54 ) 

55 

56 handler.setFormatter(formatter) 

57 root_logger.addHandler(handler) 

58 

59 # Reduce noise from third-party libraries 

60 logging.getLogger("httpx").setLevel(logging.WARNING) 

61 logging.getLogger("httpcore").setLevel(logging.WARNING) 

62 logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING) 

63 

64 

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

66 """Get a logger instance with the given name. 

67 

68 Args: 

69 name: Logger name (typically __name__) 

70 

71 Returns: 

72 Logger instance 

73 """ 

74 return logging.getLogger(name) 

75 

76 

77def log_slow_query( 

78 query: str, 

79 duration_ms: float, 

80 params: Any = None, 

81) -> None: 

82 """Log a slow query at WARNING level with structured fields. 

83 

84 Args: 

85 query: The SQL query text. 

86 duration_ms: Query duration in milliseconds. 

87 params: Optional query parameters (sanitized before logging). 

88 """ 

89 sanitized_params = params 

90 if params is not None and not isinstance(params, (str, bytes)): 

91 if isinstance(params, dict): 

92 sanitized_params = {k: _trunc_log_value(v) for k, v in params.items()} 

93 elif isinstance(params, (list, tuple)): 

94 sanitized_params = [_trunc_log_value(v) for v in params] 

95 

96 extra = { 

97 "query_type": "slow_query", 

98 "duration_ms": round(duration_ms, 2), 

99 "query": str(query)[:500], 

100 "params": sanitized_params, 

101 } 

102 logger = get_logger("slow_query") 

103 logger.warning( 

104 "Slow query (%.1f ms): %s", 

105 duration_ms, 

106 query[:200], 

107 extra=extra, 

108 ) 

109 

110 

111def log_api_request( 

112 method: str, 

113 path: str, 

114 status_code: int, 

115 duration_ms: float, 

116 request_id: str | None = None, 

117) -> None: 

118 """Log an API request at INFO level with structured fields. 

119 

120 Args: 

121 method: HTTP method (GET, POST, etc.). 

122 path: Request path. 

123 status_code: HTTP response status code. 

124 duration_ms: Request duration in milliseconds. 

125 request_id: Optional request identifier for traceability. 

126 """ 

127 extra: dict[str, Any] = { 

128 "query_type": "api_request", 

129 "http_method": method.upper(), 

130 "path": path, 

131 "status_code": status_code, 

132 "duration_ms": round(duration_ms, 2), 

133 } 

134 if request_id: 

135 extra["request_id"] = request_id 

136 

137 logger = get_logger("api") 

138 level = logging.WARNING if status_code >= 500 else logging.INFO 

139 logger.log( 

140 level, 

141 "%s %s -> %d (%.1f ms)", 

142 method.upper(), 

143 path, 

144 status_code, 

145 duration_ms, 

146 extra=extra, 

147 ) 

148 

149 

150def _trunc_log_value(value: Any, max_len: int = 100) -> Any: 

151 """Truncate a value for safe log output.""" 

152 s = str(value) 

153 return s[:max_len] + "..." if len(s) > max_len else s