Coverage for src / ezpl / handlers / file.py: 71.43%

1# /////////////////////////////////////////////////////////////// 

2# EZPL - File Logger Handler 

3# Project: ezpl 

4# /////////////////////////////////////////////////////////////// 

5 

6""" 

7File logger handler for Ezpl logging framework. 

8 

9This module provides a file-based logging handler with advanced formatting, 

10session separation, and structured output. 

11""" 

12 

13from __future__ import annotations 

14 

15# /////////////////////////////////////////////////////////////// 

16# IMPORTS 

17# /////////////////////////////////////////////////////////////// 

18# Standard library imports 

19from datetime import datetime 

20from pathlib import Path 

21from typing import Any 

22 

23# Third-party imports 

24from loguru import logger 

25from loguru._logger import Logger as LoguruLogger 

26 

27# Local imports 

28from ..core.exceptions import FileOperationError, LoggingError, ValidationError 

29from ..core.interfaces import LoggingHandler 

30from ..types.enums import LogLevel 

31from ..utils import safe_str_convert, sanitize_for_file 

32 

33# /////////////////////////////////////////////////////////////// 

34# CLASSES 

35# /////////////////////////////////////////////////////////////// 

36 

37 

38class EzLogger(LoggingHandler): 

39 """ 

40 File logger handler with advanced formatting and session management. 

41 

42 This handler provides file-based logging with: 

43 - Structured log format 

44 - Session separators 

45 - HTML tag sanitization 

46 - Automatic file creation 

47 """ 

48 

49 # /////////////////////////////////////////////////////////////// 

50 # INIT 

51 # /////////////////////////////////////////////////////////////// 

52 

53 def __init__( 

54 self, 

55 log_file: Path | str, 

56 level: str = "INFO", 

57 rotation: str | None = None, 

58 retention: str | None = None, 

59 compression: str | None = None, 

60 ) -> None: 

61 """ 

62 Initialize the file logger handler. 

63 

64 Args: 

65 log_file: Path to the log file 

66 level: The desired logging level 

67 rotation: Rotation size (e.g., "10 MB") or time (e.g., "1 day") 

68 retention: Retention period (e.g., "7 days") 

69 compression: Compression format (e.g., "zip", "gz") 

70 

71 Raises: 

72 ValidationError: If the provided level is invalid 

73 FileOperationError: If file operations fail 

74 """ 

75 if not LogLevel.is_valid_level(level): 

76 raise ValidationError(f"Invalid log level: {level}", "level", level) 

77 

78 self._level = level.upper() 

79 self._level_manually_set = False 

80 self._log_file = Path(log_file) 

81 self._logger = logger.bind(task="logger") 

82 self._logger_id: int | None = None 

83 self._rotation = rotation 

84 self._retention = retention 

85 self._compression = compression 

86 

87 # Validate and create parent directory 

88 try: 

89 self._log_file.parent.mkdir(parents=True, exist_ok=True) 

90 except (PermissionError, OSError) as e: 

91 raise FileOperationError( 

92 f"Cannot create log directory: {e}", 

93 str(self._log_file.parent), 

94 "create_directory", 

95 ) from e 

96 

97 # Validate that the file can be created/written 

98 try: 

99 if not self._log_file.exists(): 

100 self._log_file.touch() 

101 # Write test 

102 with open(self._log_file, "a", encoding="utf-8") as f: 

103 f.write("") 

104 except (PermissionError, OSError) as e: 

105 raise FileOperationError( 

106 f"Cannot write to log file: {e}", str(self._log_file), "write" 

107 ) from e 

108 

109 self._initialize_logger() 

110 

111 # ------------------------------------------------ 

112 # PRIVATE HELPER METHODS 

113 # ------------------------------------------------ 

114 

115 def _initialize_logger(self) -> None: 

116 """ 

117 Initialize the file logger handler. 

118 

119 Raises: 

120 LoggingError: If logger initialization fails 

121 """ 

122 try: 

123 # Remove existing handler if any 

124 logger_id: int | None = self._logger_id 

125 if logger_id is not None: 

126 self._logger.remove(logger_id) 

127 

128 # Call loguru.add() with keyword arguments directly 

129 # Note: loguru.add() accepts keyword arguments, not a dict 

130 self._logger_id = self._logger.add( 

131 sink=self._log_file, 

132 level=self._level, 

133 format=self._custom_formatter, # type: ignore[arg-type] 

134 filter=lambda record: record["extra"]["task"] == "logger", 

135 encoding="utf-8", 

136 rotation=self._rotation if self._rotation else None, 

137 retention=self._retention if self._retention else None, 

138 compression=self._compression if self._compression else None, 

139 ) 

140 except Exception as e: 

141 raise LoggingError(f"Failed to initialize file logger: {e}", "file") from e 

142 

143 # /////////////////////////////////////////////////////////////// 

144 # UTILS METHODS 

145 # /////////////////////////////////////////////////////////////// 

146 

147 @property 

148 def level(self) -> str: 

149 """Return the current logging level.""" 

150 return self._level 

151 

152 @property 

153 def level_manually_set(self) -> bool: 

154 """Return whether level was set manually at runtime.""" 

155 return self._level_manually_set 

156 

157 @property 

158 def rotation(self) -> str | None: 

159 """Return current rotation setting.""" 

160 return self._rotation 

161 

162 @property 

163 def retention(self) -> str | None: 

164 """Return current retention setting.""" 

165 return self._retention 

166 

167 @property 

168 def compression(self) -> str | None: 

169 """Return current compression setting.""" 

170 return self._compression 

171 

172 def mark_level_as_configured(self) -> None: 

173 """Mark the current level as coming from configuration (not manual set).""" 

174 self._level_manually_set = False 

175 

176 def set_level(self, level: str) -> None: 

177 """ 

178 Set the logging level. 

179 

180 Args: 

181 level: The desired logging level 

182 

183 Raises: 

184 ValidationError: If the provided level is invalid 

185 LoggingError: If level update fails 

186 """ 

187 if not LogLevel.is_valid_level(level): 

188 raise ValidationError(f"Invalid log level: {level}", "level", level) 

189 

190 old_level = self._level 

191 try: 

192 self._level = level.upper() 

193 self._level_manually_set = True 

194 self._initialize_logger() 

195 except Exception as e: 

196 self._level = old_level # Rollback to previous level on failure 

197 raise LoggingError(f"Failed to update log level: {e}", "file") from e 

198 

199 def log(self, level: str, message: Any) -> None: 

200 """ 

201 Log a message with the specified level. 

202 

203 Args: 

204 level: The log level 

205 message: The message to log (any type, will be converted to string) 

206 

207 Raises: 

208 ValidationError: If the level is invalid 

209 LoggingError: If logging fails 

210 """ 

211 if not LogLevel.is_valid_level(level): 

212 raise ValidationError(f"Invalid log level: {level}", "level", level) 

213 

214 # Convert message to string robustly 

215 message = safe_str_convert(message) 

216 

217 try: 

218 log_method = getattr(self._logger, level.lower()) 

219 log_method(message) 

220 except Exception as e: 

221 raise LoggingError(f"Failed to log message: {e}", "file") from e 

222 

223 # /////////////////////////////////////////////////////////////// 

224 # LOGGING METHODS (API primaire - delegates to loguru) 

225 # /////////////////////////////////////////////////////////////// 

226 

227 def trace(self, message: Any, *args, **kwargs) -> None: 

228 """Log a trace message.""" 

229 message = safe_str_convert(message) 

230 self._logger.trace(message, *args, **kwargs) 

231 

232 def debug(self, message: Any, *args, **kwargs) -> None: 

233 """Log a debug message.""" 

234 message = safe_str_convert(message) 

235 self._logger.debug(message, *args, **kwargs) 

236 

237 def info(self, message: Any, *args, **kwargs) -> None: 

238 """Log an info message.""" 

239 message = safe_str_convert(message) 

240 self._logger.info(message, *args, **kwargs) 

241 

242 def success(self, message: Any, *args, **kwargs) -> None: 

243 """Log a success message.""" 

244 message = safe_str_convert(message) 

245 self._logger.success(message, *args, **kwargs) 

246 

247 def warning(self, message: Any, *args, **kwargs) -> None: 

248 """Log a warning message.""" 

249 message = safe_str_convert(message) 

250 self._logger.warning(message, *args, **kwargs) 

251 

252 def warn(self, message: Any, *args, **kwargs) -> None: 

253 """Alias for warning(). Log a warning message.""" 

254 self.warning(message, *args, **kwargs) 

255 

256 def error(self, message: Any, *args, **kwargs) -> None: 

257 """Log an error message.""" 

258 message = safe_str_convert(message) 

259 self._logger.error(message, *args, **kwargs) 

260 

261 def critical(self, message: Any, *args, **kwargs) -> None: 

262 """Log a critical message.""" 

263 message = safe_str_convert(message) 

264 self._logger.critical(message, *args, **kwargs) 

265 

266 def exception(self, message: Any, *args, **kwargs) -> None: 

267 """Log an exception with traceback.""" 

268 message = safe_str_convert(message) 

269 self._logger.exception(message, *args, **kwargs) 

270 

271 # /////////////////////////////////////////////////////////////// 

272 # LOGURU-SPECIFIC METHODS (delegation) 

273 # /////////////////////////////////////////////////////////////// 

274 

275 def bind(self, **kwargs: Any) -> Any: 

276 """Bind context variables to the logger.""" 

277 return self._logger.bind(**kwargs) 

278 

279 def opt(self, **kwargs: Any) -> Any: 

280 """Configure logger options.""" 

281 return self._logger.opt(**kwargs) 

282 

283 def patch(self, patcher: Any) -> Any: 

284 """Patch log records.""" 

285 return self._logger.patch(patcher) 

286 

287 # /////////////////////////////////////////////////////////////// 

288 # GETTER - Returns the underlying loguru logger for advanced usage 

289 # /////////////////////////////////////////////////////////////// 

290 

291 def get_loguru(self) -> LoguruLogger: 

292 """ 

293 Get the underlying Loguru logger instance for advanced usage. 

294 

295 **Returns:** 

296 

297 * loguru.Logger: The loguru logger instance 

298 

299 **Raises:** 

300 

301 * LoggingError: If the logger is not initialized 

302 """ 

303 if not self._logger: 303 ↛ 304line 303 didn't jump to line 304 because the condition on line 303 was never true

304 raise LoggingError("File logger not initialized", "file") 

305 return self._logger # type: ignore[return-value] 

306 

307 def get_log_file(self) -> Path: 

308 """ 

309 Get the current log file path. 

310 

311 Returns: 

312 Path to the log file 

313 """ 

314 return self._log_file 

315 

316 def get_file_size(self) -> int: 

317 """ 

318 Get the current log file size in bytes. 

319 

320 Returns: 

321 File size in bytes, or 0 if file doesn't exist or error occurs 

322 """ 

323 try: 

324 if self._log_file.exists(): 324 ↛ 326line 324 didn't jump to line 326 because the condition on line 324 was always true

325 return self._log_file.stat().st_size 

326 return 0 

327 except Exception: 

328 return 0 

329 

330 def close(self) -> None: 

331 """ 

332 Close the logger handler and release file handles. 

333 

334 This method removes the loguru handler to release file handles, 

335 which is especially important on Windows where files can remain locked. 

336 """ 

337 try: 

338 # Remove existing handler if any 

339 logger_id: int | None = self._logger_id 

340 if logger_id is not None: 

341 # loguru.remove() synchronously flushes and closes the file handle 

342 self._logger.remove(logger_id) 

343 self._logger_id = None 

344 except Exception as e: 

345 raise LoggingError("Failed to close logger", "file") from e 

346 

347 # /////////////////////////////////////////////////////////////// 

348 # FILE OPERATIONS 

349 # /////////////////////////////////////////////////////////////// 

350 

351 def add_separator(self) -> None: 

352 """ 

353 Add a separator line to the log file for session distinction. 

354 

355 Raises: 

356 FileOperationError: If writing to the log file fails 

357 """ 

358 try: 

359 current_time = datetime.now().strftime("%Y-%m-%d - %H:%M") 

360 separator = f"\n\n## ==> {current_time}\n## /////////////////////////////////////////////////////////////////\n" 

361 with open(self._log_file, "a", encoding="utf-8") as log_file: 

362 log_file.write(separator) 

363 except Exception as e: 

364 raise FileOperationError( 

365 f"Failed to add separator to log file: {e}", 

366 str(self._log_file), 

367 "write", 

368 ) from e 

369 

370 # /////////////////////////////////////////////////////////////// 

371 # FORMATTING METHODS 

372 # /////////////////////////////////////////////////////////////// 

373 

374 def _custom_formatter(self, record: dict[str, Any]) -> str: 

375 """ 

376 Custom formatter for file output. 

377 

378 Args: 

379 record: Loguru record to format 

380 

381 Returns: 

382 Formatted log message (always returns a string, never raises an exception) 

383 """ 

384 try: 

385 level = ( 

386 record.get("level", {}).name 

387 if hasattr(record.get("level", {}), "name") 

388 else "INFO" 

389 ) 

390 log_level = LogLevel[level] 

391 return self._format_message(record, log_level) 

392 except Exception as e: 

393 # Never raise an exception inside a formatter — return a safe error message 

394 try: 

395 return f"????-??-?? ??:??:?? | FORMAT_ERR | unknown:unknown:? - [FORMAT ERROR: {type(e).__name__}]\n" 

396 except Exception: 

397 return "????-??-?? ??:??:?? | FORMAT_ERR | unknown:unknown:? - [FORMAT ERROR]\n" 

398 

399 def _format_message(self, record: dict[str, Any], log_level: LogLevel) -> str: 

400 """ 

401 Format a log message for file output. 

402 

403 Args: 

404 record: Loguru record 

405 log_level: LogLevel enum instance 

406 

407 Returns: 

408 Formatted log message (always returns a valid string) 

409 """ 

410 try: 

411 # Safely format the timestamp 

412 try: 

413 time_obj: Any = record.get("time") 

414 # Check if time_obj is a datetime-like object with strftime 

415 if time_obj is not None: 415 ↛ 423line 415 didn't jump to line 423 because the condition on line 415 was always true

416 strftime_method = getattr(time_obj, "strftime", None) 

417 if strftime_method is not None and callable(strftime_method): 417 ↛ 421line 417 didn't jump to line 421 because the condition on line 417 was always true

418 # Safe to call strftime - time_obj is datetime-like 

419 timestamp = strftime_method("%Y-%m-%d %H:%M:%S") 

420 else: 

421 timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") 

422 else: 

423 timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") 

424 except Exception: 

425 timestamp = "????-??-?? ??:??:??" 

426 

427 # Clean the message robustly 

428 message = safe_str_convert(record.get("message", "")) 

429 # Sanitize for file output (removes problematic characters) 

430 message = sanitize_for_file(message) 

431 

432 # Clean the function name 

433 fn = str(record.get("function", "unknown")) 

434 fn = fn.replace("<", "").replace(">", "") 

435 

436 # Safely extract module and line 

437 module = str(record.get("module", "unknown")) 

438 line = str(record.get("line", "?")) 

439 

440 return ( 

441 f"{timestamp} | " 

442 f"{log_level.label:<10} | " 

443 f"{module}:{fn}:{line} - " 

444 f"{message}\n" 

445 ) 

446 except Exception as e: 

447 # Safe fallback 

448 try: 

449 return f"????-??-?? ??:??:?? | FORMAT_ERR | unknown:unknown:? - [FORMAT ERROR: {type(e).__name__}]\n" 

450 except Exception: 

451 return "????-??-?? ??:??:?? | FORMAT_ERR | unknown:unknown:? - [FORMAT ERROR]\n" 

452 

453 # /////////////////////////////////////////////////////////////// 

454 # REPRESENTATION METHODS 

455 # /////////////////////////////////////////////////////////////// 

456 

457 def __str__(self) -> str: 

458 """String representation of the file logger.""" 

459 return f"EzLogger(file={self._log_file}, level={self._level})" 

460 

461 def __repr__(self) -> str: 

462 """Detailed string representation of the file logger.""" 

463 return f"EzLogger(file={self._log_file}, level={self._level}, logger_id={self._logger_id})"