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

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, # 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 # logger.bind() returns a BoundLogger internally; cast to declared return type 

306 return cast(LoguruLogger, self._logger) 

307 

308 def get_log_file(self) -> Path: 

309 """ 

310 Get the current log file path. 

311 

312 Returns: 

313 Path to the log file 

314 """ 

315 return self._log_file 

316 

317 def get_file_size(self) -> int: 

318 """ 

319 Get the current log file size in bytes. 

320 

321 Returns: 

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

323 """ 

324 try: 

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

326 return self._log_file.stat().st_size 

327 return 0 

328 except Exception: 

329 return 0 

330 

331 def close(self) -> None: 

332 """ 

333 Close the logger handler and release file handles. 

334 

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

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

337 """ 

338 try: 

339 # Remove existing handler if any 

340 logger_id: int | None = self._logger_id 

341 if logger_id is not None: 

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

343 self._logger.remove(logger_id) 

344 self._logger_id = None 

345 except Exception as e: 

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

347 

348 # /////////////////////////////////////////////////////////////// 

349 # FILE OPERATIONS 

350 # /////////////////////////////////////////////////////////////// 

351 

352 def add_separator(self) -> None: 

353 """ 

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

355 

356 Raises: 

357 FileOperationError: If writing to the log file fails 

358 """ 

359 try: 

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

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

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

363 log_file.write(separator) 

364 except Exception as e: 

365 raise FileOperationError( 

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

367 str(self._log_file), 

368 "write", 

369 ) from e 

370 

371 # /////////////////////////////////////////////////////////////// 

372 # FORMATTING METHODS 

373 # /////////////////////////////////////////////////////////////// 

374 

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

376 """ 

377 Custom formatter for file output. 

378 

379 Args: 

380 record: Loguru record to format 

381 

382 Returns: 

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

384 """ 

385 try: 

386 level = ( 

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

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

389 else "INFO" 

390 ) 

391 log_level = LogLevel[level] 

392 return self._format_message(record, log_level) 

393 except Exception as e: 

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

395 try: 

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

397 except Exception: 

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

399 

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

401 """ 

402 Format a log message for file output. 

403 

404 Args: 

405 record: Loguru record 

406 log_level: LogLevel enum instance 

407 

408 Returns: 

409 Formatted log message (always returns a valid string) 

410 """ 

411 try: 

412 # Safely format the timestamp 

413 try: 

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

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

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

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

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

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

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

421 else: 

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

423 else: 

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

425 except Exception: 

426 timestamp = "????-??-?? ??:??:??" 

427 

428 # Clean the message robustly 

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

430 # Sanitize for file output (removes problematic characters) 

431 message = sanitize_for_file(message) 

432 

433 # Clean the function name 

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

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

436 

437 # Safely extract module and line 

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

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

440 

441 return ( 

442 f"{timestamp} | " 

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

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

445 f"{message}\n" 

446 ) 

447 except Exception as e: 

448 # Safe fallback 

449 try: 

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

451 except Exception: 

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

453 

454 # /////////////////////////////////////////////////////////////// 

455 # REPRESENTATION METHODS 

456 # /////////////////////////////////////////////////////////////// 

457 

458 def __str__(self) -> str: 

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

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

461 

462 def __repr__(self) -> str: 

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

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

465 

466 

467# /////////////////////////////////////////////////////////////// 

468# PUBLIC API 

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

470 

471__all__ = [ 

472 "EzLogger", 

473]