Coverage for src/lexigram/graphql/core/error_logging.py: 99%

1"""Structured error logging for GraphQL. 

2 

3This module provides structured logging for GraphQL errors 

4to support debugging and monitoring. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass, field 

10from datetime import UTC, datetime 

11import traceback 

12from typing import Any 

13 

14from lexigram.logging import get_logger 

15 

16logger = get_logger(__name__) 

17 

18 

19@dataclass 

20class ErrorLogEntry: 

21 """Structured error log entry. 

22 

23 Attributes: 

24 timestamp: When the error occurred. 

25 error_type: Type of the error. 

26 message: Error message. 

27 query: GraphQL query (may be masked). 

28 variables: Query variables (may be masked). 

29 path: Path in the GraphQL document. 

30 extensions: Error extensions. 

31 stacktrace: Exception stacktrace. 

32 context: Additional context. 

33 """ 

34 

35 timestamp: datetime = field(default_factory=lambda: datetime.now(UTC)) 

36 error_type: str = "" 

37 message: str = "" 

38 query: str | None = None 

39 variables: dict[str, Any] | None = None 

40 path: list[str] | None = None 

41 extensions: dict[str, Any] = field(default_factory=dict) 

42 stacktrace: str | None = None 

43 context: dict[str, Any] = field(default_factory=dict) 

44 

45 def to_dict(self) -> dict[str, Any]: 

46 """Convert to dictionary. 

47 

48 Returns: 

49 Dictionary representation. 

50 """ 

51 return { 

52 "timestamp": self.timestamp.isoformat(), 

53 "error_type": self.error_type, 

54 "message": self.message, 

55 "query": self.query, 

56 "variables": self.variables, 

57 "path": self.path, 

58 "extensions": self.extensions, 

59 "stacktrace": self.stacktrace, 

60 "context": self.context, 

61 } 

62 

63 

64class ErrorLogger: 

65 """Structured error logger for GraphQL. 

66 

67 Example: 

68 ```python 

69 logger = ErrorLogger() 

70 

71 # Log an error 

72 logger.log_error( 

73 error=exc, 

74 query="{ user { name } }", 

75 variables={"id": "1"}, 

76 ) 

77 ``` 

78 """ 

79 

80 def __init__( 

81 self, 

82 log_queries: bool = True, 

83 log_variables: bool = False, 

84 log_stacktrace: bool = True, 

85 max_query_length: int = 500, 

86 ): 

87 """Initialize the error logger. 

88 

89 Args: 

90 log_queries: Whether to log queries. 

91 log_variables: Whether to log query variables. 

92 log_stacktrace: Whether to log stack traces. 

93 max_query_length: Maximum query length to log. 

94 """ 

95 self._log_queries = log_queries 

96 self._log_variables = log_variables 

97 self._log_stacktrace = log_stacktrace 

98 self._max_query_length = max_query_length 

99 

100 def log_error( 

101 self, 

102 error: Exception, 

103 query: str | None = None, 

104 variables: dict[str, Any] | None = None, 

105 path: list[str] | None = None, 

106 extensions: dict[str, Any] | None = None, 

107 context: dict[str, Any] | None = None, 

108 ) -> ErrorLogEntry: 

109 """Log a GraphQL error. 

110 

111 Args: 

112 error: The exception that occurred. 

113 query: The GraphQL query. 

114 variables: Query variables. 

115 path: Path in the GraphQL document. 

116 extensions: Error extensions. 

117 context: Additional context. 

118 

119 Returns: 

120 The created log entry. 

121 """ 

122 # Build the entry 

123 entry = ErrorLogEntry( 

124 error_type=type(error).__name__, 

125 message=str(error), 

126 path=path, 

127 extensions=extensions or {}, 

128 context=context or {}, 

129 ) 

130 

131 # Mask query if needed 

132 if query and self._log_queries: 

133 entry.query = query[: self._max_query_length] 

134 

135 # Mask variables if needed 

136 if variables and self._log_variables: 

137 entry.variables = variables 

138 elif variables: 

139 # Mask sensitive variables 

140 entry.variables = self._mask_variables(variables) 

141 

142 # Add stacktrace if needed 

143 if self._log_stacktrace: 

144 entry.stacktrace = traceback.format_exc() 

145 

146 # Log the error 

147 logger.error( 

148 "GraphQL error: %s: %s", 

149 entry.error_type, 

150 entry.message, 

151 extra=entry.to_dict(), 

152 ) 

153 

154 return entry 

155 

156 def _mask_variables(self, variables: dict[str, Any]) -> dict[str, Any]: 

157 """Mask sensitive variables. 

158 

159 Args: 

160 variables: Variables to mask. 

161 

162 Returns: 

163 Masked variables. 

164 """ 

165 sensitive_keys = {"password", "secret", "token", "api_key", "credit_card"} 

166 

167 masked = {} 

168 for key, value in variables.items(): 

169 if any(s in key.lower() for s in sensitive_keys): 

170 masked[key] = "***REDACTED***" 

171 else: 

172 masked[key] = value 

173 

174 return masked 

175 

176 

177class QueryLogger: 

178 """Structured logger for GraphQL queries. 

179 

180 Logs queries with timing and metadata for monitoring. 

181 """ 

182 

183 def __init__(self, enabled: bool = True): 

184 """Initialize the query logger. 

185 

186 Args: 

187 enabled: Whether logging is enabled. 

188 """ 

189 self._enabled = enabled 

190 

191 def log_query( 

192 self, 

193 query: str, 

194 variables: dict | None = None, 

195 operation_name: str | None = None, 

196 duration_ms: float | None = None, 

197 errors_count: int = 0, 

198 ) -> None: 

199 """Log a GraphQL query. 

200 

201 Args: 

202 query: The GraphQL query. 

203 variables: Query variables. 

204 operation_name: Name of the operation. 

205 duration_ms: Query duration in milliseconds. 

206 errors_count: Number of errors. 

207 """ 

208 if not self._enabled: 

209 return 

210 

211 # Determine operation type 

212 operation_type = self._extract_operation_type(query) 

213 

214 # Build log data 

215 log_data = { 

216 "query": query[:500], # Truncate for logging 

217 "operation_type": operation_type, 

218 "operation_name": operation_name, 

219 "variables": variables, 

220 "duration_ms": duration_ms, 

221 "errors_count": errors_count, 

222 } 

223 

224 # Log based on errors 

225 if errors_count > 0: 

226 logger.warning( 

227 "GraphQL query completed with errors: %s, duration: %sms", 

228 operation_type, 

229 duration_ms, 

230 extra=log_data, 

231 ) 

232 else: 

233 logger.info( 

234 "GraphQL query completed: %s, duration: %sms", 

235 operation_type, 

236 duration_ms, 

237 extra=log_data, 

238 ) 

239 

240 def _extract_operation_type(self, query: str) -> str: 

241 """Extract the operation type from a query. 

242 

243 Args: 

244 query: GraphQL query string. 

245 

246 Returns: 

247 Operation type (query, mutation, subscription). 

248 """ 

249 query_lower = query.strip().lower() 

250 

251 if query_lower.startswith("mutation"): 

252 return "mutation" 

253 if query_lower.startswith("subscription"): 

254 return "subscription" 

255 return "query" 

256 

257 

258# Default logger instance 

259default_error_logger = ErrorLogger() 

260default_query_logger = QueryLogger() 

261 

262 

263__all__ = [ 

264 "ErrorLogEntry", 

265 "ErrorLogger", 

266 "QueryLogger", 

267 "default_error_logger", 

268 "default_query_logger", 

269]