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

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, cast 

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, 

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 error(self, message: Any, *args, **kwargs) -> None: 

253 """Log an error message.""" 

254 message = safe_str_convert(message) 

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

256 

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

258 """Log a critical message.""" 

259 message = safe_str_convert(message) 

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

261 

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

263 """Log an exception with traceback.""" 

264 message = safe_str_convert(message) 

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

266 

267 # /////////////////////////////////////////////////////////////// 

268 # LOGURU-SPECIFIC METHODS (delegation) 

269 # /////////////////////////////////////////////////////////////// 

270 

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

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

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

274 

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

276 """Configure logger options.""" 

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

278 

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

280 """Patch log records.""" 

281 return self._logger.patch(patcher) 

282 

283 # /////////////////////////////////////////////////////////////// 

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

285 # /////////////////////////////////////////////////////////////// 

286 

287 def get_loguru(self) -> LoguruLogger: 

288 """ 

289 Get the underlying Loguru logger instance for advanced usage. 

290 

291 **Returns:** 

292 

293 * loguru.Logger: The loguru logger instance 

294 

295 **Raises:** 

296 

297 * LoggingError: If the logger is not initialized 

298 """ 

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

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

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

302 return cast(LoguruLogger, self._logger) 

303 

304 def get_log_file(self) -> Path: 

305 """ 

306 Get the current log file path. 

307 

308 Returns: 

309 Path to the log file 

310 """ 

311 return self._log_file 

312 

313 def get_file_size(self) -> int: 

314 """ 

315 Get the current log file size in bytes. 

316 

317 Returns: 

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

319 """ 

320 try: 

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

322 return self._log_file.stat().st_size 

323 return 0 

324 except Exception: 

325 return 0 

326 

327 def close(self) -> None: 

328 """ 

329 Close the logger handler and release file handles. 

330 

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

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

333 """ 

334 try: 

335 # Remove existing handler if any 

336 logger_id: int | None = self._logger_id 

337 if logger_id is not None: 

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

339 self._logger.remove(logger_id) 

340 self._logger_id = None 

341 except Exception as e: 

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

343 

344 # /////////////////////////////////////////////////////////////// 

345 # FILE OPERATIONS 

346 # /////////////////////////////////////////////////////////////// 

347 

348 def add_separator(self) -> None: 

349 """ 

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

351 

352 Raises: 

353 FileOperationError: If writing to the log file fails 

354 """ 

355 try: 

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

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

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

359 log_file.write(separator) 

360 except Exception as e: 

361 raise FileOperationError( 

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

363 str(self._log_file), 

364 "write", 

365 ) from e 

366 

367 # /////////////////////////////////////////////////////////////// 

368 # FORMATTING METHODS 

369 # /////////////////////////////////////////////////////////////// 

370 

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

372 """ 

373 Custom formatter for file output. 

374 

375 Args: 

376 record: Loguru record to format 

377 

378 Returns: 

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

380 """ 

381 try: 

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

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

384 

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})" 

464 

465 

466# /////////////////////////////////////////////////////////////// 

467# PUBLIC API 

468# /////////////////////////////////////////////////////////////// 

469 

470__all__ = [ 

471 "EzLogger", 

472]