Coverage for stlog/output.py: 82%

1from __future__ import annotations 

2 

3import logging 

4import logging.handlers 

5import os 

6import sys 

7import typing 

8from dataclasses import dataclass, field 

9 

10from stlog.base import ( 

11 GLOBAL_LOGGING_CONFIG, 

12 StlogError, 

13 check_env_false, 

14) 

15from stlog.filter import ContextReinjectFilter 

16from stlog.formatter import ( 

17 Formatter, 

18 HumanFormatter, 

19 RichHumanFormatter, 

20) 

21from stlog.handler import CustomRichHandler 

22 

23RICH_INSTALLED: bool = False 

24try: 

25 from rich.console import Console 

26 

27 RICH_INSTALLED = True 

28except ImportError: 

29 pass 

30 

31 

32def _get_default_use_rich() -> bool | None: 

33 tmp = os.environ.get("STLOG_USE_RICH") 

34 if tmp is None: 

35 return None 

36 if tmp.strip().upper() in ("NONE", "AUTO", ""): 

37 return None 

38 return tmp.strip().upper() in ("1", "TRUE", "YES") 

39 

40 

41DEFAULT_USE_RICH = _get_default_use_rich() 

42 

43 

44@dataclass 

45class Output: 

46 """Abstract output base class. 

47 

48 Attributes: 

49 formatter: the Python logging Formatter to use. 

50 level: python logging level specific to this output 

51 (None means "use the global logging level"). 

52 filters: list of logging Filters (or simple callables) to filter some LogRecord 

53 for this specific output. 

54 reinject_context_in_standard_logging: if True, reinject the LogContext 

55 in log record emitted with python standard loggers 

56 (note: override `stlog.setup` default value for this output). 

57 read_extra_kwargs_from_standard_logging: if try to reinject the extra kwargs from standard logging 

58 (note: override `stlog.setup` default value for this output). 

59 

60 """ 

61 

62 _handler: logging.Handler = field(init=False, default_factory=logging.NullHandler) 

63 formatter: logging.Formatter | None = None 

64 level: int | str | None = None 

65 filters: typing.Iterable[ 

66 typing.Callable[[logging.LogRecord], bool] | logging.Filter 

67 ] = field(default_factory=list) 

68 reinject_context_in_standard_logging: bool | None = None 

69 read_extra_kwargs_from_standard_logging: bool | None = None 

70 

71 @property 

72 def _reinject_context_in_standard_logging(self) -> bool: 

73 # lazy evaluation because Output are built before setup() call 

74 if self.reinject_context_in_standard_logging is not None: 

75 return self.reinject_context_in_standard_logging 

76 if GLOBAL_LOGGING_CONFIG.reinject_context_in_standard_logging is not None: 

77 return GLOBAL_LOGGING_CONFIG.reinject_context_in_standard_logging 

78 return check_env_false("STLOG_REINJECT_CONTEXT_IN_STANDARD_LOGGING", True) 

79 

80 @property 

81 def _read_extra_kwargs_from_standard_logging(self) -> bool: 

82 # lazy evaluation because Output are built before setup() call 

83 if self.read_extra_kwargs_from_standard_logging is not None: 

84 return self.read_extra_kwargs_from_standard_logging 

85 if GLOBAL_LOGGING_CONFIG.read_extra_kwargs_from_standard_logging is not None: 

86 return GLOBAL_LOGGING_CONFIG.read_extra_kwargs_from_standard_logging 

87 return check_env_false("STLOG_READ_EXTRA_KWARGS_FROM_STANDARD_LOGGING", True) 

88 

89 def set_handler( 

90 self, 

91 handler: logging.Handler, 

92 ): 

93 """Configure the Python logging Handler to use.""" 

94 self._handler = handler 

95 self._handler.setFormatter(self.get_formatter_or_raise()) 

96 if self.level is not None: 

97 self._handler.setLevel(self.level) 

98 if self._reinject_context_in_standard_logging: 

99 self._handler.addFilter( 

100 ContextReinjectFilter( 

101 read_extra_kwargs_from_standard_logging=self._read_extra_kwargs_from_standard_logging 

102 ) 

103 ) 

104 for filter in self.filters: 

105 self._handler.addFilter(filter) 

106 

107 def get_handler(self) -> logging.Handler: 

108 """Get the configured Python logging Handler.""" 

109 return self._handler 

110 

111 def get_formatter_or_raise(self) -> logging.Formatter: 

112 if self.formatter is None: 

113 raise StlogError("formatter is not set") 

114 return self.formatter 

115 

116 

117@dataclass 

118class StreamOutput(Output): 

119 """Represent an output to a stream (stdout, stderr...). 

120 

121 Attributes: 

122 stream: the stream to use (`typing.TextIO`), default to `sys.stderr`. 

123 

124 """ 

125 

126 stream: typing.TextIO = sys.stderr 

127 

128 def __post_init__(self): 

129 if self.formatter is None: 

130 self.formatter = HumanFormatter() 

131 self.set_handler( 

132 logging.StreamHandler(self.stream), 

133 ) 

134 

135 

136@dataclass 

137class RichStreamOutput(StreamOutput): 

138 force_terminal: bool = False 

139 

140 def __post_init__(self): 

141 if not RICH_INSTALLED: 

142 raise StlogError("Rich is not installed and RichStreamOutput is specified") 

143 if self.formatter is None: 

144 self.formatter = RichHumanFormatter() 

145 self.set_handler( 

146 CustomRichHandler(stream=self.stream, force_terminal=self.force_terminal) 

147 ) 

148 

149 

150def make_stream_or_rich_stream_output( 

151 stream: typing.TextIO = sys.stderr, 

152 use_rich: bool | None = DEFAULT_USE_RICH, 

153 rich_formatter: Formatter | None = None, 

154 not_rich_formatter: Formatter | None = None, 

155 **kwargs, 

156) -> StreamOutput: 

157 """Create automatically a `stlog.output.RichStreamOutput` or a (classic)`stlog.output.StreamOutput`. 

158 

159 To get a `stlog.output.RichStreamOutput`, following conditions must be true: 

160 

161 - `rich` library must be installed and available in python path 

162 - `use_rich` parameter must be `True` (forced mode) or `None` (automatic mode) 

163 - (if `use_rich` is `None`): the selected `stream` must "output" in a real terminal (not in a shell filter 

164 or in a file through redirection...) 

165 

166 NOTE: the default value of the `use_rich` parameter is `None` (automatic) but it can be forced by 

167 the `STLOG_USE_RICH` env variable. 

168 

169 Attributes: 

170 stream: the stream to use (`typing.TextIO`), default to `sys.stderr`. 

171 use_rich: if None, use [rich output](https://github.com/Textualize/rich/blob/master/README.md) if possible 

172 (rich installed and supported tty), if True/False force the usage (or not). 

173 rich_formatter: Formatter to use if rich is available/selected (None => default RichHumanFormatter instance). 

174 not_rich_formatter: Formatter to use if rich is not available/selected (None => default HumanFormatter instance). 

175 

176 """ 

177 if "formatter" in kwargs: 

178 raise StlogError( 

179 "you can't use formatter in kwargs for this function (buy you can use rich_formatter/not_rich_formatter instead)" 

180 ) 

181 _use_rich: bool = False 

182 if use_rich is not None: 

183 # manual mode 

184 _use_rich = use_rich 

185 # automatic mode 

186 elif RICH_INSTALLED: 

187 c = Console(file=stream) 

188 _use_rich = c.is_terminal 

189 if _use_rich: 

190 return RichStreamOutput( 

191 stream=stream, 

192 formatter=rich_formatter or RichHumanFormatter(), 

193 **kwargs, 

194 ) 

195 else: 

196 return StreamOutput( 

197 stream=stream, 

198 formatter=not_rich_formatter or HumanFormatter(), 

199 **kwargs, 

200 ) 

201 

202 

203@dataclass 

204class FileOutput(Output): 

205 """Represent an output to a file. 

206 

207 Attributes: 

208 filename: the filename to use. 

209 mode: the mode to use, default to "a". 

210 encoding: the encoding to use, default to None. 

211 delay: if True, the file is not opened until the first call to emit(). 

212 errors: the errors to use, default to None (python >= 3.9 only) 

213 

214 """ 

215 

216 filename: str = "" 

217 mode: str = "a" 

218 encoding: str | None = None 

219 delay: bool = False 

220 errors: str | None = None 

221 

222 def __post_init__(self): 

223 if not self.filename: 

224 raise StlogError("filename is not set") 

225 if self.formatter is None: 

226 self.formatter = HumanFormatter() 

227 kwargs = { 

228 "mode": self.mode, 

229 "encoding": self.encoding, 

230 "delay": self.delay, 

231 } 

232 if sys.version_info >= (3, 9): 

233 kwargs["errors"] = self.errors 

234 self.set_handler( 

235 logging.FileHandler(self.filename, **kwargs), # type: ignore 

236 ) 

237 

238 

239@dataclass 

240class RotatingFileOutput(FileOutput): 

241 """Represent an output to a rotating file. 

242 

243 Attributes: 

244 filename: the filename to use. 

245 mode: the mode to use, default to "a". 

246 max_bytes: the maximum number of bytes to use, default to 0. 

247 backup_count: the number of backup files to use, default to 0. 

248 encoding: the encoding to use, default to None. 

249 delay: if True, the file is not opened until the first call to emit(). 

250 errors: the errors to use, default to None. 

251 

252 """ 

253 

254 max_bytes: int = 0 

255 backup_count: int = 0 

256 

257 def __post_init__(self): 

258 if not self.filename: 

259 raise StlogError("filename is not set") 

260 if self.formatter is None: 

261 self.formatter = HumanFormatter() 

262 kwargs = { 

263 "mode": self.mode, 

264 "encoding": self.encoding, 

265 "delay": self.delay, 

266 "backupCount": self.backup_count, 

267 "maxBytes": self.max_bytes, 

268 } 

269 if sys.version_info >= (3, 9): 

270 kwargs["errors"] = self.errors 

271 self.set_handler( 

272 logging.handlers.RotatingFileHandler( 

273 self.filename, 

274 **kwargs, # type: ignore 

275 ), 

276 )