Coverage for src/ezplog/handlers/file.py: 75.00%
192 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-10-16 19:04 +0000
1# ///////////////////////////////////////////////////////////////
2# EZPL - File Logger Handler
3# Project: ezpl
4# ///////////////////////////////////////////////////////////////
6"""
7File logger handler for Ezpl logging framework.
9This module provides a file-based logging handler with advanced formatting,
10session separation, and structured output.
11"""
13from __future__ import annotations
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
24# Third-party imports
25from loguru import logger
26from loguru._logger import Logger as LoguruLogger
28# Local imports
29from ..core.exceptions import FileOperationError, LoggingError, ValidationError
30from ..core.interfaces import LoggingHandler
31from ..types.aliases import (
32 SUPPORTED_COMPRESSIONS,
33 CompressionSpec,
34 RetentionSpec,
35 RotationSpec,
36)
37from ..types.enums import LogLevel
38from ..utils import safe_str_convert, sanitize_for_file
40# ///////////////////////////////////////////////////////////////
41# CLASSES
42# ///////////////////////////////////////////////////////////////
45class EzLogger(LoggingHandler):
46 """
47 File logger handler with advanced formatting and session management.
49 This handler provides file-based logging with:
50 - Structured log format
51 - Session separators
52 - HTML tag sanitization
53 - Automatic file creation
54 """
56 # ///////////////////////////////////////////////////////////////
57 # INIT
58 # ///////////////////////////////////////////////////////////////
60 def __init__(
61 self,
62 log_file: Path | str,
63 level: str = "INFO",
64 rotation: RotationSpec = None,
65 retention: RetentionSpec = None,
66 compression: CompressionSpec = None,
67 *,
68 backtrace: bool = True,
69 diagnose: bool = False,
70 ) -> None:
71 """
72 Initialize the file logger handler.
74 Args:
75 log_file: Path to the log file
76 level: The desired logging level
77 rotation: Rotation trigger — size string ("10 MB"), byte count,
78 datetime.time, datetime.timedelta, or a predicate callable
79 retention: Retention policy — duration string ("7 days"), number of
80 files to keep (int), timedelta, or a cleanup callable
81 compression: Compression format name (see SUPPORTED_COMPRESSIONS)
82 or a callable applied to the rotated file path
83 backtrace: If True, extend tracebacks beyond the catching point
84 diagnose: If True, include local variable values in tracebacks.
85 Defaults to False, unlike loguru: locals routinely hold
86 secrets, and log files are archived and shipped elsewhere.
88 Raises:
89 ValidationError: If the provided level is invalid
90 FileOperationError: If file operations fail
91 """
92 if not LogLevel.is_valid_level(level):
93 raise ValidationError(f"Invalid log level: {level}", "level", level)
95 if isinstance(compression, str) and compression not in SUPPORTED_COMPRESSIONS:
96 raise ValidationError(
97 f"Unsupported compression format: {compression!r}. "
98 f"Expected one of: {', '.join(sorted(SUPPORTED_COMPRESSIONS))}",
99 "compression",
100 compression,
101 )
103 self._level = level.upper()
104 self._level_manually_set = False
105 self._log_file = Path(log_file)
106 self._logger = logger.bind(task="logger")
107 self._logger_id: int | None = None
108 self._rotation: RotationSpec = rotation
109 self._retention: RetentionSpec = retention
110 self._compression: CompressionSpec = compression
111 self._backtrace = backtrace
112 self._diagnose = diagnose
114 # Validate and create parent directory
115 try:
116 self._log_file.parent.mkdir(parents=True, exist_ok=True)
117 except OSError as e:
118 raise FileOperationError(
119 f"Cannot create log directory: {e}",
120 str(self._log_file.parent),
121 "create_directory",
122 ) from e
124 # Validate that the file can be created/written
125 try:
126 if not self._log_file.exists():
127 self._log_file.touch()
128 # Write test
129 with open(self._log_file, "a", encoding="utf-8") as f:
130 f.write("")
131 except OSError as e:
132 raise FileOperationError(
133 f"Cannot write to log file: {e}", str(self._log_file), "write"
134 ) from e
136 self._initialize_logger()
138 # ------------------------------------------------
139 # PRIVATE HELPER METHODS
140 # ------------------------------------------------
142 def _initialize_logger(self) -> None:
143 """
144 Initialize the file logger handler.
146 Raises:
147 LoggingError: If logger initialization fails
148 """
149 try:
150 # Remove existing handler if any
151 logger_id: int | None = self._logger_id
152 if logger_id is not None:
153 self._logger.remove(logger_id)
154 else:
155 # First initialization: drop loguru's default stderr sink so
156 # intercepted stdlib records only go to the file, not the console.
157 with suppress(ValueError):
158 logger.remove(0)
160 # Call loguru.add() with keyword arguments directly
161 # Note: loguru.add() accepts keyword arguments, not a dict
162 self._logger_id = self._logger.add(
163 sink=self._log_file,
164 level=self._level,
165 format=self._custom_formatter,
166 filter=lambda record: record["extra"]["task"] == "logger",
167 encoding="utf-8",
168 rotation=self._rotation if self._rotation else None,
169 retention=self._retention if self._retention else None,
170 compression=self._compression if self._compression else None,
171 backtrace=self._backtrace,
172 diagnose=self._diagnose,
173 )
174 except Exception as e:
175 raise LoggingError(f"Failed to initialize file logger: {e}", "file") from e
177 # ///////////////////////////////////////////////////////////////
178 # UTILS METHODS
179 # ///////////////////////////////////////////////////////////////
181 @property
182 def level(self) -> str:
183 """Return the current logging level."""
184 return self._level
186 @property
187 def level_manually_set(self) -> bool:
188 """Return whether level was set manually at runtime."""
189 return self._level_manually_set
191 @property
192 def rotation(self) -> RotationSpec:
193 """Return current rotation setting."""
194 return self._rotation
196 @property
197 def retention(self) -> RetentionSpec:
198 """Return current retention setting."""
199 return self._retention
201 @property
202 def compression(self) -> CompressionSpec:
203 """Return current compression setting."""
204 return self._compression
206 @property
207 def backtrace(self) -> bool:
208 """Return whether tracebacks extend beyond the catching point."""
209 return self._backtrace
211 @property
212 def diagnose(self) -> bool:
213 """Return whether local variable values appear in tracebacks."""
214 return self._diagnose
216 def mark_level_as_configured(self) -> None:
217 """Mark the current level as coming from configuration (not manual set)."""
218 self._level_manually_set = False
220 def set_level(self, level: str) -> None:
221 """
222 Set the logging level.
224 Args:
225 level: The desired logging level
227 Raises:
228 ValidationError: If the provided level is invalid
229 LoggingError: If level update fails
230 """
231 if not LogLevel.is_valid_level(level):
232 raise ValidationError(f"Invalid log level: {level}", "level", level)
234 old_level = self._level
235 try:
236 self._level = level.upper()
237 self._level_manually_set = True
238 self._initialize_logger()
239 except Exception as e:
240 self._level = old_level # Rollback to previous level on failure
241 raise LoggingError(f"Failed to update log level: {e}", "file") from e
243 def log(self, level: str, message: Any) -> None:
244 """
245 Log a message with the specified level.
247 Args:
248 level: The log level
249 message: The message to log (any type, will be converted to string)
251 Raises:
252 ValidationError: If the level is invalid
253 LoggingError: If logging fails
254 """
255 if not LogLevel.is_valid_level(level):
256 raise ValidationError(f"Invalid log level: {level}", "level", level)
258 # Convert message to string robustly
259 message = safe_str_convert(message)
261 try:
262 log_method = getattr(self._logger, level.lower())
263 log_method(message)
264 except Exception as e:
265 raise LoggingError(f"Failed to log message: {e}", "file") from e
267 # ///////////////////////////////////////////////////////////////
268 # LOGGING METHODS (API primaire - delegates to loguru)
269 # ///////////////////////////////////////////////////////////////
271 # NOTE: every wrapper below uses `opt(depth=1)` so loguru attributes the
272 # record to the actual caller (user code) rather than to this method.
273 # Without it, every log line reports `file:info:<line>` etc.
275 def trace(self, message: Any, *args, **kwargs) -> None:
276 """Log a trace message."""
277 message = safe_str_convert(message)
278 self._logger.opt(depth=1).trace(message, *args, **kwargs)
280 def debug(self, message: Any, *args, **kwargs) -> None:
281 """Log a debug message."""
282 message = safe_str_convert(message)
283 self._logger.opt(depth=1).debug(message, *args, **kwargs)
285 def info(self, message: Any, *args, **kwargs) -> None:
286 """Log an info message."""
287 message = safe_str_convert(message)
288 self._logger.opt(depth=1).info(message, *args, **kwargs)
290 def success(self, message: Any, *args, **kwargs) -> None:
291 """Log a success message."""
292 message = safe_str_convert(message)
293 self._logger.opt(depth=1).success(message, *args, **kwargs)
295 def warning(self, message: Any, *args, **kwargs) -> None:
296 """Log a warning message."""
297 message = safe_str_convert(message)
298 self._logger.opt(depth=1).warning(message, *args, **kwargs)
300 def error(self, message: Any, *args, **kwargs) -> None:
301 """Log an error message."""
302 message = safe_str_convert(message)
303 self._logger.opt(depth=1).error(message, *args, **kwargs)
305 def critical(self, message: Any, *args, **kwargs) -> None:
306 """Log a critical message."""
307 message = safe_str_convert(message)
308 self._logger.opt(depth=1).critical(message, *args, **kwargs)
310 def exception(self, message: Any, *args, **kwargs) -> None:
311 """Log an exception with traceback."""
312 message = safe_str_convert(message)
313 self._logger.opt(depth=1, exception=True).log("ERROR", message, *args, **kwargs)
315 # ///////////////////////////////////////////////////////////////
316 # LOGURU-SPECIFIC METHODS (delegation)
317 # ///////////////////////////////////////////////////////////////
319 def bind(self, **kwargs: Any) -> Any:
320 """Bind context variables to the logger."""
321 return self._logger.bind(**kwargs)
323 def opt(self, **kwargs: Any) -> Any:
324 """Configure logger options."""
325 return self._logger.opt(**kwargs)
327 def patch(self, patcher: Any) -> Any:
328 """Patch log records."""
329 return self._logger.patch(patcher)
331 # ///////////////////////////////////////////////////////////////
332 # GETTER - Returns the underlying loguru logger for advanced usage
333 # ///////////////////////////////////////////////////////////////
335 def get_loguru(self) -> LoguruLogger:
336 """
337 Get the underlying Loguru logger instance for advanced usage.
339 **Returns:**
341 * loguru.Logger: The loguru logger instance
343 **Raises:**
345 * LoggingError: If the logger is not initialized
346 """
347 if not self._logger: 347 ↛ 348line 347 didn't jump to line 348 because the condition on line 347 was never true
348 raise LoggingError("File logger not initialized", "file")
349 # logger.bind() returns a BoundLogger internally; cast to declared return type
350 return cast(LoguruLogger, self._logger)
352 def get_log_file(self) -> Path:
353 """
354 Get the current log file path.
356 Returns:
357 Path to the log file
358 """
359 return self._log_file
361 def get_file_size(self) -> int:
362 """
363 Get the current log file size in bytes.
365 Returns:
366 File size in bytes, or 0 if file doesn't exist or error occurs
367 """
368 try:
369 if self._log_file.exists(): 369 ↛ 371line 369 didn't jump to line 371 because the condition on line 369 was always true
370 return self._log_file.stat().st_size
371 return 0
372 except Exception:
373 return 0
375 def close(self) -> None:
376 """
377 Close the logger handler and release file handles.
379 This method removes the loguru handler to release file handles,
380 which is especially important on Windows where files can remain locked.
381 """
382 try:
383 # Remove existing handler if any
384 logger_id: int | None = self._logger_id
385 if logger_id is not None:
386 # loguru.remove() synchronously flushes and closes the file handle
387 self._logger.remove(logger_id)
388 self._logger_id = None
389 except Exception as e:
390 raise LoggingError("Failed to close logger", "file") from e
392 # ///////////////////////////////////////////////////////////////
393 # FILE OPERATIONS
394 # ///////////////////////////////////////////////////////////////
396 def add_separator(self) -> None:
397 """
398 Add a separator line to the log file for session distinction.
400 Raises:
401 FileOperationError: If writing to the log file fails
402 """
403 try:
404 current_time = datetime.now().strftime("%Y-%m-%d - %H:%M")
405 separator = f"\n\n## ==> {current_time}\n## /////////////////////////////////////////////////////////////////\n"
406 with open(self._log_file, "a", encoding="utf-8") as log_file:
407 log_file.write(separator)
408 except Exception as e:
409 raise FileOperationError(
410 f"Failed to add separator to log file: {e}",
411 str(self._log_file),
412 "write",
413 ) from e
415 # ///////////////////////////////////////////////////////////////
416 # FORMATTING METHODS
417 # ///////////////////////////////////////////////////////////////
419 def _custom_formatter(self, record: Any) -> str:
420 """
421 Custom formatter for file output.
423 Args:
424 record: Loguru record to format
426 Returns:
427 Formatted log message (always returns a string, never raises an exception)
428 """
429 try:
430 if not isinstance(record, dict): 430 ↛ 431line 430 didn't jump to line 431 because the condition on line 430 was never true
431 return "????-??-?? ??:??:?? | FORMAT_ERR | unknown:unknown:? - [FORMAT ERROR: InvalidRecord]\n"
433 level = (
434 record.get("level", {}).name
435 if hasattr(record.get("level", {}), "name")
436 else "INFO"
437 )
438 log_level = LogLevel[level]
439 return self._format_message(record, log_level)
440 except Exception as e:
441 # Never raise an exception inside a formatter — return a safe error message
442 try:
443 return f"????-??-?? ??:??:?? | FORMAT_ERR | unknown:unknown:? - [FORMAT ERROR: {type(e).__name__}]\n"
444 except Exception:
445 return "????-??-?? ??:??:?? | FORMAT_ERR | unknown:unknown:? - [FORMAT ERROR]\n"
447 def _format_message(self, record: dict[str, Any], log_level: LogLevel) -> str:
448 """
449 Format a log message for file output.
451 Args:
452 record: Loguru record
453 log_level: LogLevel enum instance
455 Returns:
456 Formatted log message (always returns a valid string)
457 """
458 try:
459 # Safely format the timestamp
460 try:
461 time_obj: Any = record.get("time")
462 # Check if time_obj is a datetime-like object with strftime
463 if time_obj is not None: 463 ↛ 471line 463 didn't jump to line 471 because the condition on line 463 was always true
464 strftime_method = getattr(time_obj, "strftime", None)
465 if strftime_method is not None and callable(strftime_method): 465 ↛ 469line 465 didn't jump to line 469 because the condition on line 465 was always true
466 # Safe to call strftime - time_obj is datetime-like
467 timestamp = strftime_method("%Y-%m-%d %H:%M:%S")
468 else:
469 timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
470 else:
471 timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
472 except Exception:
473 timestamp = "????-??-?? ??:??:??"
475 # Clean the message robustly
476 message = safe_str_convert(record.get("message", ""))
477 # Sanitize for file output (removes problematic characters)
478 message = sanitize_for_file(message)
480 # Clean the function name
481 fn = str(record.get("function", "unknown"))
482 fn = fn.replace("<", "").replace(">", "")
484 # Safely extract module and line
485 module = str(record.get("module", "unknown"))
486 line = str(record.get("line", "?"))
488 # NOTE: loguru applies `.format_map(record)` on the value returned by a
489 # callable formatter, so any literal `{` / `}` in the interpolated content
490 # (e.g. boto3 debug payloads like {'Bucket': ...}) would be re-parsed as
491 # placeholders -> KeyError / "Max string recursion exceeded". Doubling
492 # braces neutralizes them since nothing here needs to remain a placeholder.
493 formatted = (
494 f"{timestamp} | "
495 f"{log_level.label:<10} | "
496 f"{module}:{fn}:{line} - "
497 f"{message}\n"
498 )
499 return formatted.replace("{", "{{").replace("}", "}}")
500 except Exception as e:
501 # Safe fallback
502 try:
503 return f"????-??-?? ??:??:?? | FORMAT_ERR | unknown:unknown:? - [FORMAT ERROR: {type(e).__name__}]\n"
504 except Exception:
505 return "????-??-?? ??:??:?? | FORMAT_ERR | unknown:unknown:? - [FORMAT ERROR]\n"
507 # ///////////////////////////////////////////////////////////////
508 # REPRESENTATION METHODS
509 # ///////////////////////////////////////////////////////////////
511 def __str__(self) -> str:
512 """String representation of the file logger."""
513 return f"EzLogger(file={self._log_file}, level={self._level})"
515 def __repr__(self) -> str:
516 """Detailed string representation of the file logger."""
517 return f"EzLogger(file={self._log_file}, level={self._level}, logger_id={self._logger_id})"
520# ///////////////////////////////////////////////////////////////
521# PUBLIC API
522# ///////////////////////////////////////////////////////////////
524__all__ = [
525 "EzLogger",
526]