Coverage for src/ezplog/handlers/file.py: 71.92%

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 contextlib import suppress 

20from datetime import datetime 

21from pathlib import Path 

22from typing import Any, cast 

23 

24# Third-party imports 

25from loguru import logger 

26from loguru._logger import Logger as LoguruLogger 

27 

28# Local imports 

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

30from ..core.interfaces import LoggingHandler 

31from ..types.enums import LogLevel 

32from ..utils import safe_str_convert, sanitize_for_file 

33 

34# /////////////////////////////////////////////////////////////// 

35# CLASSES 

36# /////////////////////////////////////////////////////////////// 

37 

38 

39class EzLogger(LoggingHandler): 

40 """ 

41 File logger handler with advanced formatting and session management. 

42 

43 This handler provides file-based logging with: 

44 - Structured log format 

45 - Session separators 

46 - HTML tag sanitization 

47 - Automatic file creation 

48 """ 

49 

50 # /////////////////////////////////////////////////////////////// 

51 # INIT 

52 # /////////////////////////////////////////////////////////////// 

53 

54 def __init__( 

55 self, 

56 log_file: Path | str, 

57 level: str = "INFO", 

58 rotation: str | None = None, 

59 retention: str | None = None, 

60 compression: str | None = None, 

61 ) -> None: 

62 """ 

63 Initialize the file logger handler. 

64 

65 Args: 

66 log_file: Path to the log file 

67 level: The desired logging level 

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

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

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

71 

72 Raises: 

73 ValidationError: If the provided level is invalid 

74 FileOperationError: If file operations fail 

75 """ 

76 if not LogLevel.is_valid_level(level): 

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

78 

79 self._level = level.upper() 

80 self._level_manually_set = False 

81 self._log_file = Path(log_file) 

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

83 self._logger_id: int | None = None 

84 self._rotation = rotation 

85 self._retention = retention 

86 self._compression = compression 

87 

88 # Validate and create parent directory 

89 try: 

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

91 except OSError as e: 

92 raise FileOperationError( 

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

94 str(self._log_file.parent), 

95 "create_directory", 

96 ) from e 

97 

98 # Validate that the file can be created/written 

99 try: 

100 if not self._log_file.exists(): 

101 self._log_file.touch() 

102 # Write test 

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

104 f.write("") 

105 except OSError as e: 

106 raise FileOperationError( 

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

108 ) from e 

109 

110 self._initialize_logger() 

111 

112 # ------------------------------------------------ 

113 # PRIVATE HELPER METHODS 

114 # ------------------------------------------------ 

115 

116 def _initialize_logger(self) -> None: 

117 """ 

118 Initialize the file logger handler. 

119 

120 Raises: 

121 LoggingError: If logger initialization fails 

122 """ 

123 try: 

124 # Remove existing handler if any 

125 logger_id: int | None = self._logger_id 

126 if logger_id is not None: 

127 self._logger.remove(logger_id) 

128 else: 

129 # First initialization: drop loguru's default stderr sink so 

130 # intercepted stdlib records only go to the file, not the console. 

131 with suppress(ValueError): 

132 logger.remove(0) 

133 

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

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

136 self._logger_id = self._logger.add( 

137 sink=self._log_file, 

138 level=self._level, 

139 format=self._custom_formatter, 

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

141 encoding="utf-8", 

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

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

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

145 ) 

146 except Exception as e: 

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

148 

149 # /////////////////////////////////////////////////////////////// 

150 # UTILS METHODS 

151 # /////////////////////////////////////////////////////////////// 

152 

153 @property 

154 def level(self) -> str: 

155 """Return the current logging level.""" 

156 return self._level 

157 

158 @property 

159 def level_manually_set(self) -> bool: 

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

161 return self._level_manually_set 

162 

163 @property 

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

165 """Return current rotation setting.""" 

166 return self._rotation 

167 

168 @property 

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

170 """Return current retention setting.""" 

171 return self._retention 

172 

173 @property 

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

175 """Return current compression setting.""" 

176 return self._compression 

177 

178 def mark_level_as_configured(self) -> None: 

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

180 self._level_manually_set = False 

181 

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

183 """ 

184 Set the logging level. 

185 

186 Args: 

187 level: The desired logging level 

188 

189 Raises: 

190 ValidationError: If the provided level is invalid 

191 LoggingError: If level update fails 

192 """ 

193 if not LogLevel.is_valid_level(level): 

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

195 

196 old_level = self._level 

197 try: 

198 self._level = level.upper() 

199 self._level_manually_set = True 

200 self._initialize_logger() 

201 except Exception as e: 

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

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

204 

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

206 """ 

207 Log a message with the specified level. 

208 

209 Args: 

210 level: The log level 

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

212 

213 Raises: 

214 ValidationError: If the level is invalid 

215 LoggingError: If logging fails 

216 """ 

217 if not LogLevel.is_valid_level(level): 

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

219 

220 # Convert message to string robustly 

221 message = safe_str_convert(message) 

222 

223 try: 

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

225 log_method(message) 

226 except Exception as e: 

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

228 

229 # /////////////////////////////////////////////////////////////// 

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

231 # /////////////////////////////////////////////////////////////// 

232 

233 # NOTE: every wrapper below uses `opt(depth=1)` so loguru attributes the 

234 # record to the actual caller (user code) rather than to this method. 

235 # Without it, every log line reports `file:info:<line>` etc. 

236 

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

238 """Log a trace message.""" 

239 message = safe_str_convert(message) 

240 self._logger.opt(depth=1).trace(message, *args, **kwargs) 

241 

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

243 """Log a debug message.""" 

244 message = safe_str_convert(message) 

245 self._logger.opt(depth=1).debug(message, *args, **kwargs) 

246 

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

248 """Log an info message.""" 

249 message = safe_str_convert(message) 

250 self._logger.opt(depth=1).info(message, *args, **kwargs) 

251 

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

253 """Log a success message.""" 

254 message = safe_str_convert(message) 

255 self._logger.opt(depth=1).success(message, *args, **kwargs) 

256 

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

258 """Log a warning message.""" 

259 message = safe_str_convert(message) 

260 self._logger.opt(depth=1).warning(message, *args, **kwargs) 

261 

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

263 """Log an error message.""" 

264 message = safe_str_convert(message) 

265 self._logger.opt(depth=1).error(message, *args, **kwargs) 

266 

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

268 """Log a critical message.""" 

269 message = safe_str_convert(message) 

270 self._logger.opt(depth=1).critical(message, *args, **kwargs) 

271 

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

273 """Log an exception with traceback.""" 

274 message = safe_str_convert(message) 

275 self._logger.opt(depth=1, exception=True).log("ERROR", message, *args, **kwargs) 

276 

277 # /////////////////////////////////////////////////////////////// 

278 # LOGURU-SPECIFIC METHODS (delegation) 

279 # /////////////////////////////////////////////////////////////// 

280 

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

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

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

284 

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

286 """Configure logger options.""" 

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

288 

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

290 """Patch log records.""" 

291 return self._logger.patch(patcher) 

292 

293 # /////////////////////////////////////////////////////////////// 

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

295 # /////////////////////////////////////////////////////////////// 

296 

297 def get_loguru(self) -> LoguruLogger: 

298 """ 

299 Get the underlying Loguru logger instance for advanced usage. 

300 

301 **Returns:** 

302 

303 * loguru.Logger: The loguru logger instance 

304 

305 **Raises:** 

306 

307 * LoggingError: If the logger is not initialized 

308 """ 

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

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

311 # logger.bind() returns a BoundLogger internally; cast to declared return type 

312 return cast(LoguruLogger, self._logger) 

313 

314 def get_log_file(self) -> Path: 

315 """ 

316 Get the current log file path. 

317 

318 Returns: 

319 Path to the log file 

320 """ 

321 return self._log_file 

322 

323 def get_file_size(self) -> int: 

324 """ 

325 Get the current log file size in bytes. 

326 

327 Returns: 

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

329 """ 

330 try: 

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

332 return self._log_file.stat().st_size 

333 return 0 

334 except Exception: 

335 return 0 

336 

337 def close(self) -> None: 

338 """ 

339 Close the logger handler and release file handles. 

340 

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

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

343 """ 

344 try: 

345 # Remove existing handler if any 

346 logger_id: int | None = self._logger_id 

347 if logger_id is not None: 

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

349 self._logger.remove(logger_id) 

350 self._logger_id = None 

351 except Exception as e: 

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

353 

354 # /////////////////////////////////////////////////////////////// 

355 # FILE OPERATIONS 

356 # /////////////////////////////////////////////////////////////// 

357 

358 def add_separator(self) -> None: 

359 """ 

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

361 

362 Raises: 

363 FileOperationError: If writing to the log file fails 

364 """ 

365 try: 

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

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

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

369 log_file.write(separator) 

370 except Exception as e: 

371 raise FileOperationError( 

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

373 str(self._log_file), 

374 "write", 

375 ) from e 

376 

377 # /////////////////////////////////////////////////////////////// 

378 # FORMATTING METHODS 

379 # /////////////////////////////////////////////////////////////// 

380 

381 def _custom_formatter(self, record: Any) -> str: 

382 """ 

383 Custom formatter for file output. 

384 

385 Args: 

386 record: Loguru record to format 

387 

388 Returns: 

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

390 """ 

391 try: 

392 if not isinstance(record, dict): 392 ↛ 393line 392 didn't jump to line 393 because the condition on line 392 was never true

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

394 

395 level = ( 

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

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

398 else "INFO" 

399 ) 

400 log_level = LogLevel[level] 

401 return self._format_message(record, log_level) 

402 except Exception as e: 

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

404 try: 

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

406 except Exception: 

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

408 

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

410 """ 

411 Format a log message for file output. 

412 

413 Args: 

414 record: Loguru record 

415 log_level: LogLevel enum instance 

416 

417 Returns: 

418 Formatted log message (always returns a valid string) 

419 """ 

420 try: 

421 # Safely format the timestamp 

422 try: 

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

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

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

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

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

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

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

430 else: 

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

432 else: 

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

434 except Exception: 

435 timestamp = "????-??-?? ??:??:??" 

436 

437 # Clean the message robustly 

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

439 # Sanitize for file output (removes problematic characters) 

440 message = sanitize_for_file(message) 

441 

442 # Clean the function name 

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

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

445 

446 # Safely extract module and line 

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

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

449 

450 # NOTE: loguru applies `.format_map(record)` on the value returned by a 

451 # callable formatter, so any literal `{` / `}` in the interpolated content 

452 # (e.g. boto3 debug payloads like {'Bucket': ...}) would be re-parsed as 

453 # placeholders -> KeyError / "Max string recursion exceeded". Doubling 

454 # braces neutralizes them since nothing here needs to remain a placeholder. 

455 formatted = ( 

456 f"{timestamp} | " 

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

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

459 f"{message}\n" 

460 ) 

461 return formatted.replace("{", "{{").replace("}", "}}") 

462 except Exception as e: 

463 # Safe fallback 

464 try: 

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

466 except Exception: 

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

468 

469 # /////////////////////////////////////////////////////////////// 

470 # REPRESENTATION METHODS 

471 # /////////////////////////////////////////////////////////////// 

472 

473 def __str__(self) -> str: 

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

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

476 

477 def __repr__(self) -> str: 

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

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

480 

481 

482# /////////////////////////////////////////////////////////////// 

483# PUBLIC API 

484# /////////////////////////////////////////////////////////////// 

485 

486__all__ = [ 

487 "EzLogger", 

488]