Coverage for muutils / logger / loggingstream.py: 71%

1from __future__ import annotations 

2 

3import sys 

4import time 

5from dataclasses import dataclass, field 

6from typing import Any, Callable 

7 

8if sys.version_info >= (3, 12): 

9 from typing import override 

10else: 

11 from typing_extensions import override 

12 

13from muutils.logger.simplelogger import AnyIO, NullIO 

14from muutils.misc import sanitize_fname 

15 

16 

17@dataclass 

18class LoggingStream: 

19 """properties of a logging stream 

20 

21 - `name: str` name of the stream 

22 - `aliases: set[str]` aliases for the stream 

23 (calls to these names will be redirected to this stream. duplicate alises will result in errors) 

24 TODO: perhaps duplicate alises should result in duplicate writes? 

25 - `file: str|bool|AnyIO|None` file to write to 

26 - if `None`, will write to standard log 

27 - if `True`, will write to `name + ".log"` 

28 - if `False` will "write" to `NullIO` (throw it away) 

29 - if a string, will write to that file 

30 - if a fileIO type object, will write to that object 

31 - `default_level: int|None` default level for this stream 

32 - `default_contents: dict[str, Callable[[], Any]]` default contents for this stream 

33 - `last_msg: tuple[float, Any]|None` last message written to this stream (timestamp, message) 

34 """ 

35 

36 name: str | None 

37 aliases: set[str | None] = field(default_factory=set) 

38 file: str | bool | AnyIO | None = None 

39 default_level: int | None = None 

40 default_contents: dict[str, Callable[[], Any]] = field(default_factory=dict) 

41 handler: AnyIO | None = None 

42 

43 # TODO: implement last-message caching 

44 # last_msg: tuple[float, Any]|None = None 

45 

46 def make_handler(self) -> AnyIO | None: 

47 if self.file is None: 

48 return None 

49 elif isinstance(self.file, str): 

50 # if its a string, open a file 

51 return open( 

52 self.file, 

53 "w", 

54 encoding="utf-8", 

55 ) 

56 elif isinstance(self.file, bool): 

57 # if its a bool and true, open a file with the same name as the stream (in the current dir) 

58 # TODO: make this happen in the same dir as the main logfile? 

59 if self.file: 

60 return open( # type: ignore[return-value] 

61 f"{sanitize_fname(self.name)}.log.jsonl", 

62 "w", 

63 encoding="utf-8", 

64 ) 

65 else: 

66 return NullIO() 

67 else: 

68 # if its neither, check it has `.write()` and `.flush()` methods 

69 if ( 

70 ( 

71 not hasattr(self.file, "write") 

72 or (not callable(self.file.write)) 

73 or (not hasattr(self.file, "flush")) 

74 or (not callable(self.file.flush)) 

75 ) 

76 or (not hasattr(self.file, "close")) 

77 or (not callable(self.file.close)) 

78 ): 

79 raise ValueError(f"stream {self.name} has invalid handler {self.file}") 

80 # ignore type check because we know it has a .write() method, 

81 # assume the user knows what they're doing 

82 return self.file # type: ignore 

83 

84 def __post_init__(self): 

85 self.aliases = set(self.aliases) 

86 if any(x.startswith("_") for x in self.aliases if x is not None): 

87 raise ValueError( 

88 "stream names or aliases cannot start with an underscore, sorry" 

89 ) 

90 self.aliases.add(self.name) 

91 self.default_contents["_timestamp"] = time.time 

92 self.default_contents["_stream"] = lambda: self.name 

93 self.handler = self.make_handler() 

94 

95 def __del__(self): 

96 if self.handler is not None: 

97 self.handler.flush() 

98 self.handler.close() 

99 

100 @override 

101 def __str__(self): 

102 return f"LoggingStream(name={self.name}, aliases={self.aliases}, file={self.file}, default_level={self.default_level}, default_contents={self.default_contents})"