Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sqlalchemy/log.py: 67%

1# log.py 

2# Copyright (C) 2006-2024 the SQLAlchemy authors and contributors 

3# <see AUTHORS file> 

4# Includes alterations by Vinay Sajip vinay_sajip@yahoo.co.uk 

5# 

6# This module is part of SQLAlchemy and is released under 

7# the MIT License: https://www.opensource.org/licenses/mit-license.php 

8 

9"""Logging control and utilities. 

10 

11Control of logging for SA can be performed from the regular python logging 

12module. The regular dotted module namespace is used, starting at 

13'sqlalchemy'. For class-level logging, the class name is appended. 

14 

15The "echo" keyword parameter, available on SQLA :class:`_engine.Engine` 

16and :class:`_pool.Pool` objects, corresponds to a logger specific to that 

17instance only. 

18 

19""" 

20from __future__ import annotations 

21 

22import logging 

23import sys 

24from typing import Any 

25from typing import Optional 

26from typing import overload 

27from typing import Set 

28from typing import Type 

29from typing import TypeVar 

30from typing import Union 

31 

32from .util import py311 

33from .util.typing import Literal 

34 

35 

36STACKLEVEL = True 

37# needed as of py3.11.0b1 

38# #8019 

39STACKLEVEL_OFFSET = 2 if py311 else 1 

40 

41_IT = TypeVar("_IT", bound="Identified") 

42 

43_EchoFlagType = Union[None, bool, Literal["debug"]] 

44 

45# set initial level to WARN. This so that 

46# log statements don't occur in the absence of explicit 

47# logging being enabled for 'sqlalchemy'. 

48rootlogger = logging.getLogger("sqlalchemy") 

49if rootlogger.level == logging.NOTSET: 

50 rootlogger.setLevel(logging.WARN) 

51 

52 

53def _add_default_handler(logger: logging.Logger) -> None: 

54 handler = logging.StreamHandler(sys.stdout) 

55 handler.setFormatter( 

56 logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s") 

57 ) 

58 logger.addHandler(handler) 

59 

60 

61_logged_classes: Set[Type[Identified]] = set() 

62 

63 

64def _qual_logger_name_for_cls(cls: Type[Identified]) -> str: 

65 return ( 

66 getattr(cls, "_sqla_logger_namespace", None) 

67 or cls.__module__ + "." + cls.__name__ 

68 ) 

69 

70 

71def class_logger(cls: Type[_IT]) -> Type[_IT]: 

72 logger = logging.getLogger(_qual_logger_name_for_cls(cls)) 

73 cls._should_log_debug = lambda self: logger.isEnabledFor( # type: ignore[method-assign] # noqa: E501 

74 logging.DEBUG 

75 ) 

76 cls._should_log_info = lambda self: logger.isEnabledFor( # type: ignore[method-assign] # noqa: E501 

77 logging.INFO 

78 ) 

79 cls.logger = logger 

80 _logged_classes.add(cls) 

81 return cls 

82 

83 

84_IdentifiedLoggerType = Union[logging.Logger, "InstanceLogger"] 

85 

86 

87class Identified: 

88 __slots__ = () 

89 

90 logging_name: Optional[str] = None 

91 

92 logger: _IdentifiedLoggerType 

93 

94 _echo: _EchoFlagType 

95 

96 def _should_log_debug(self) -> bool: 

97 return self.logger.isEnabledFor(logging.DEBUG) 

98 

99 def _should_log_info(self) -> bool: 

100 return self.logger.isEnabledFor(logging.INFO) 

101 

102 

103class InstanceLogger: 

104 """A logger adapter (wrapper) for :class:`.Identified` subclasses. 

105 

106 This allows multiple instances (e.g. Engine or Pool instances) 

107 to share a logger, but have its verbosity controlled on a 

108 per-instance basis. 

109 

110 The basic functionality is to return a logging level 

111 which is based on an instance's echo setting. 

112 

113 Default implementation is: 

114 

115 'debug' -> logging.DEBUG 

116 True -> logging.INFO 

117 False -> Effective level of underlying logger ( 

118 logging.WARNING by default) 

119 None -> same as False 

120 """ 

121 

122 # Map echo settings to logger levels 

123 _echo_map = { 

124 None: logging.NOTSET, 

125 False: logging.NOTSET, 

126 True: logging.INFO, 

127 "debug": logging.DEBUG, 

128 } 

129 

130 _echo: _EchoFlagType 

131 

132 __slots__ = ("echo", "logger") 

133 

134 def __init__(self, echo: _EchoFlagType, name: str): 

135 self.echo = echo 

136 self.logger = logging.getLogger(name) 

137 

138 # if echo flag is enabled and no handlers, 

139 # add a handler to the list 

140 if self._echo_map[echo] <= logging.INFO and not self.logger.handlers: 

141 _add_default_handler(self.logger) 

142 

143 # 

144 # Boilerplate convenience methods 

145 # 

146 def debug(self, msg: str, *args: Any, **kwargs: Any) -> None: 

147 """Delegate a debug call to the underlying logger.""" 

148 

149 self.log(logging.DEBUG, msg, *args, **kwargs) 

150 

151 def info(self, msg: str, *args: Any, **kwargs: Any) -> None: 

152 """Delegate an info call to the underlying logger.""" 

153 

154 self.log(logging.INFO, msg, *args, **kwargs) 

155 

156 def warning(self, msg: str, *args: Any, **kwargs: Any) -> None: 

157 """Delegate a warning call to the underlying logger.""" 

158 

159 self.log(logging.WARNING, msg, *args, **kwargs) 

160 

161 warn = warning 

162 

163 def error(self, msg: str, *args: Any, **kwargs: Any) -> None: 

164 """ 

165 Delegate an error call to the underlying logger. 

166 """ 

167 self.log(logging.ERROR, msg, *args, **kwargs) 

168 

169 def exception(self, msg: str, *args: Any, **kwargs: Any) -> None: 

170 """Delegate an exception call to the underlying logger.""" 

171 

172 kwargs["exc_info"] = 1 

173 self.log(logging.ERROR, msg, *args, **kwargs) 

174 

175 def critical(self, msg: str, *args: Any, **kwargs: Any) -> None: 

176 """Delegate a critical call to the underlying logger.""" 

177 

178 self.log(logging.CRITICAL, msg, *args, **kwargs) 

179 

180 def log(self, level: int, msg: str, *args: Any, **kwargs: Any) -> None: 

181 """Delegate a log call to the underlying logger. 

182 

183 The level here is determined by the echo 

184 flag as well as that of the underlying logger, and 

185 logger._log() is called directly. 

186 

187 """ 

188 

189 # inline the logic from isEnabledFor(), 

190 # getEffectiveLevel(), to avoid overhead. 

191 

192 if self.logger.manager.disable >= level: 

193 return 

194 

195 selected_level = self._echo_map[self.echo] 

196 if selected_level == logging.NOTSET: 

197 selected_level = self.logger.getEffectiveLevel() 

198 

199 if level >= selected_level: 

200 if STACKLEVEL: 

201 kwargs["stacklevel"] = ( 

202 kwargs.get("stacklevel", 1) + STACKLEVEL_OFFSET 

203 ) 

204 

205 self.logger._log(level, msg, args, **kwargs) 

206 

207 def isEnabledFor(self, level: int) -> bool: 

208 """Is this logger enabled for level 'level'?""" 

209 

210 if self.logger.manager.disable >= level: 

211 return False 

212 return level >= self.getEffectiveLevel() 

213 

214 def getEffectiveLevel(self) -> int: 

215 """What's the effective level for this logger?""" 

216 

217 level = self._echo_map[self.echo] 

218 if level == logging.NOTSET: 

219 level = self.logger.getEffectiveLevel() 

220 return level 

221 

222 

223def instance_logger( 

224 instance: Identified, echoflag: _EchoFlagType = None 

225) -> None: 

226 """create a logger for an instance that implements :class:`.Identified`.""" 

227 

228 if instance.logging_name: 

229 name = "%s.%s" % ( 

230 _qual_logger_name_for_cls(instance.__class__), 

231 instance.logging_name, 

232 ) 

233 else: 

234 name = _qual_logger_name_for_cls(instance.__class__) 

235 

236 instance._echo = echoflag # type: ignore 

237 

238 logger: Union[logging.Logger, InstanceLogger] 

239 

240 if echoflag in (False, None): 

241 # if no echo setting or False, return a Logger directly, 

242 # avoiding overhead of filtering 

243 logger = logging.getLogger(name) 

244 else: 

245 # if a specified echo flag, return an EchoLogger, 

246 # which checks the flag, overrides normal log 

247 # levels by calling logger._log() 

248 logger = InstanceLogger(echoflag, name) 

249 

250 instance.logger = logger # type: ignore 

251 

252 

253class echo_property: 

254 __doc__ = """\ 

255 When ``True``, enable log output for this element. 

256 

257 This has the effect of setting the Python logging level for the namespace 

258 of this element's class and object reference. A value of boolean ``True`` 

259 indicates that the loglevel ``logging.INFO`` will be set for the logger, 

260 whereas the string value ``debug`` will set the loglevel to 

261 ``logging.DEBUG``. 

262 """ 

263 

264 @overload 

265 def __get__( 

266 self, instance: Literal[None], owner: Type[Identified] 

267 ) -> echo_property: ... 

268 

269 @overload 

270 def __get__( 

271 self, instance: Identified, owner: Type[Identified] 

272 ) -> _EchoFlagType: ... 

273 

274 def __get__( 

275 self, instance: Optional[Identified], owner: Type[Identified] 

276 ) -> Union[echo_property, _EchoFlagType]: 

277 if instance is None: 

278 return self 

279 else: 

280 return instance._echo 

281 

282 def __set__(self, instance: Identified, value: _EchoFlagType) -> None: 

283 instance_logger(instance, echoflag=value)