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

1# log.py 

2# Copyright (C) 2006-2025 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 import py38 

34from .util.typing import Literal 

35 

36 

37if py38: 

38 STACKLEVEL = True 

39 # needed as of py3.11.0b1 

40 # #8019 

41 STACKLEVEL_OFFSET = 2 if py311 else 1 

42else: 

43 STACKLEVEL = False 

44 STACKLEVEL_OFFSET = 0 

45 

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

47 

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

49 

50# set initial level to WARN. This so that 

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

52# logging being enabled for 'sqlalchemy'. 

53rootlogger = logging.getLogger("sqlalchemy") 

54if rootlogger.level == logging.NOTSET: 

55 rootlogger.setLevel(logging.WARN) 

56 

57 

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

59 handler = logging.StreamHandler(sys.stdout) 

60 handler.setFormatter( 

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

62 ) 

63 logger.addHandler(handler) 

64 

65 

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

67 

68 

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

70 return ( 

71 getattr(cls, "_sqla_logger_namespace", None) 

72 or cls.__module__ + "." + cls.__name__ 

73 ) 

74 

75 

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

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

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

79 logging.DEBUG 

80 ) 

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

82 logging.INFO 

83 ) 

84 cls.logger = logger 

85 _logged_classes.add(cls) 

86 return cls 

87 

88 

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

90 

91 

92class Identified: 

93 __slots__ = () 

94 

95 logging_name: Optional[str] = None 

96 

97 logger: _IdentifiedLoggerType 

98 

99 _echo: _EchoFlagType 

100 

101 def _should_log_debug(self) -> bool: 

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

103 

104 def _should_log_info(self) -> bool: 

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

106 

107 

108class InstanceLogger: 

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

110 

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

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

113 per-instance basis. 

114 

115 The basic functionality is to return a logging level 

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

117 

118 Default implementation is: 

119 

120 'debug' -> logging.DEBUG 

121 True -> logging.INFO 

122 False -> Effective level of underlying logger ( 

123 logging.WARNING by default) 

124 None -> same as False 

125 """ 

126 

127 # Map echo settings to logger levels 

128 _echo_map = { 

129 None: logging.NOTSET, 

130 False: logging.NOTSET, 

131 True: logging.INFO, 

132 "debug": logging.DEBUG, 

133 } 

134 

135 _echo: _EchoFlagType 

136 

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

138 

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

140 self.echo = echo 

141 self.logger = logging.getLogger(name) 

142 

143 # if echo flag is enabled and no handlers, 

144 # add a handler to the list 

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

146 _add_default_handler(self.logger) 

147 

148 # 

149 # Boilerplate convenience methods 

150 # 

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

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

153 

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

155 

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

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

158 

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

160 

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

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

163 

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

165 

166 warn = warning 

167 

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

169 """ 

170 Delegate an error call to the underlying logger. 

171 """ 

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

173 

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

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

176 

177 kwargs["exc_info"] = 1 

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

179 

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

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

182 

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

184 

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

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

187 

188 The level here is determined by the echo 

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

190 logger._log() is called directly. 

191 

192 """ 

193 

194 # inline the logic from isEnabledFor(), 

195 # getEffectiveLevel(), to avoid overhead. 

196 

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

198 return 

199 

200 selected_level = self._echo_map[self.echo] 

201 if selected_level == logging.NOTSET: 

202 selected_level = self.logger.getEffectiveLevel() 

203 

204 if level >= selected_level: 

205 if STACKLEVEL: 

206 kwargs["stacklevel"] = ( 

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

208 ) 

209 

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

211 

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

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

214 

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

216 return False 

217 return level >= self.getEffectiveLevel() 

218 

219 def getEffectiveLevel(self) -> int: 

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

221 

222 level = self._echo_map[self.echo] 

223 if level == logging.NOTSET: 

224 level = self.logger.getEffectiveLevel() 

225 return level 

226 

227 

228def instance_logger( 

229 instance: Identified, echoflag: _EchoFlagType = None 

230) -> None: 

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

232 

233 if instance.logging_name: 

234 name = "%s.%s" % ( 

235 _qual_logger_name_for_cls(instance.__class__), 

236 instance.logging_name, 

237 ) 

238 else: 

239 name = _qual_logger_name_for_cls(instance.__class__) 

240 

241 instance._echo = echoflag # type: ignore 

242 

243 logger: Union[logging.Logger, InstanceLogger] 

244 

245 if echoflag in (False, None): 

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

247 # avoiding overhead of filtering 

248 logger = logging.getLogger(name) 

249 else: 

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

251 # which checks the flag, overrides normal log 

252 # levels by calling logger._log() 

253 logger = InstanceLogger(echoflag, name) 

254 

255 instance.logger = logger # type: ignore 

256 

257 

258class echo_property: 

259 __doc__ = """\ 

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

261 

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

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

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

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

266 ``logging.DEBUG``. 

267 """ 

268 

269 @overload 

270 def __get__( 

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

272 ) -> echo_property: ... 

273 

274 @overload 

275 def __get__( 

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

277 ) -> _EchoFlagType: ... 

278 

279 def __get__( 

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

281 ) -> Union[echo_property, _EchoFlagType]: 

282 if instance is None: 

283 return self 

284 else: 

285 return instance._echo 

286 

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

288 instance_logger(instance, echoflag=value)