Coverage for src / ocarina / opinionated / loggers / file_logger.py: 0.00%

1"""Logger with files generation effects.""" 

2 

3from contextlib import suppress 

4from typing import TYPE_CHECKING, final 

5 

6if TYPE_CHECKING: 

7 from pathlib import Path 

8 

9 from ocarina.custom_types.supports_write import SupportsWrite 

10 

11from ocarina.opinionated.loggers.print_logger import PrintLogger 

12 

13 

14@final 

15class FileLogger(PrintLogger): 

16 """Log with files generation effects.""" 

17 

18 def __init__( 

19 self, 

20 *, 

21 base_dir: Path, 

22 with_flush_effect: bool = True, 

23 with_fallback_on_print_logger_when_no_taxonomy_effect: bool = True, 

24 ) -> None: 

25 """Initialize the FileLogger. 

26 

27 Args: 

28 base_dir: Root directory where log files are written. 

29 with_flush_effect: Print log file content to stdout on cleanup. 

30 with_fallback_on_print_logger_when_no_taxonomy_effect: Fall back to 

31 PrintLogger when no taxonomy is set. 

32 

33 """ 

34 self._base_dir = base_dir 

35 self._with_flush_effect = with_flush_effect 

36 self._with_fallback_on_print_logger_when_no_taxonomy_effect = ( 

37 with_fallback_on_print_logger_when_no_taxonomy_effect 

38 ) 

39 super().__init__() 

40 

41 def _has_taxonomy(self) -> bool: 

42 return any(part for part in self._domain_taxonomy) 

43 

44 @property 

45 def _log_file_path(self) -> Path: 

46 if len(self._domain_taxonomy) == 1: 

47 folder_path = self._base_dir 

48 file_name = f"{self._domain_taxonomy[0]}.log" 

49 else: 

50 folder_path = self._base_dir.joinpath(*self._domain_taxonomy[:-1]) 

51 file_name = f"{self._domain_taxonomy[-1]}.log" 

52 

53 folder_path.mkdir(parents=True, exist_ok=True) 

54 return folder_path / file_name 

55 

56 def _print_log( 

57 self, 

58 prefix: str, 

59 msg: str, 

60 *args, 

61 exc: BaseException | None = None, 

62 stream: SupportsWrite[str], 

63 ) -> None: 

64 formatted_msg = msg % args if args else msg 

65 

66 output = " ".join( 

67 part 

68 for part in [ 

69 self._prefix_thunk(), 

70 prefix, 

71 formatted_msg, 

72 ] 

73 if part 

74 ) 

75 self._write_log_hook(output, exc, stream) 

76 

77 def _write_log_hook( 

78 self, 

79 output: str, 

80 exc: BaseException | None, 

81 stream: SupportsWrite[str], 

82 ) -> None: 

83 if self._has_taxonomy(): 

84 log_file_path = self._log_file_path 

85 with log_file_path.open("a", encoding="utf-8") as f: 

86 self._flush_log(output, exc, f) 

87 elif self._with_fallback_on_print_logger_when_no_taxonomy_effect: 

88 super()._write_log_hook(output, exc, stream) 

89 

90 def cleanup(self) -> None: 

91 """Remove the log file linked to the taxonomy, flushing it first.""" 

92 if self._has_taxonomy(): 

93 log_file_path = self._log_file_path 

94 

95 if log_file_path.exists(): 

96 if self._with_flush_effect: 

97 print_logger = PrintLogger().set_domain_taxonomy( 

98 self._domain_taxonomy 

99 ) 

100 print_logger.raw() 

101 print_logger.info("Flush:") 

102 with log_file_path.open("r", encoding="utf-8") as f: 

103 while chunk := f.read(1024): 

104 print_logger.raw(chunk, end="") 

105 

106 with suppress(Exception): 

107 log_file_path.unlink()