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

1# log.py 

2# Copyright (C) 2006-2026 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""" 

20 

21from __future__ import annotations 

22 

23import logging 

24import sys 

25from typing import Any 

26from typing import Literal 

27from typing import Optional 

28from typing import overload 

29from typing import Set 

30from typing import Type 

31from typing import TypeVar 

32from typing import Union 

33 

34STACKLEVEL = True 

35STACKLEVEL_OFFSET = 2 

36 

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

38 

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

40 

41# set initial level to WARN. This so that 

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

43# logging being enabled for 'sqlalchemy'. 

44rootlogger = logging.getLogger("sqlalchemy") 

45if rootlogger.level == logging.NOTSET: 

46 rootlogger.setLevel(logging.WARNING) 

47 

48 

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

50 handler = logging.StreamHandler(sys.stdout) 

51 handler.setFormatter( 

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

53 ) 

54 logger.addHandler(handler) 

55 

56 

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

58 

59 

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

61 return ( 

62 getattr(cls, "_sqla_logger_namespace", None) 

63 or cls.__module__ + "." + cls.__name__ 

64 ) 

65 

66 

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

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

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

70 logging.DEBUG 

71 ) 

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

73 logging.INFO 

74 ) 

75 cls.logger = logger 

76 _logged_classes.add(cls) 

77 return cls 

78 

79 

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

81 

82 

83class Identified: 

84 __slots__ = () 

85 

86 logging_name: Optional[str] = None 

87 

88 logger: _IdentifiedLoggerType 

89 

90 _echo: _EchoFlagType 

91 

92 def _should_log_debug(self) -> bool: 

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

94 

95 def _should_log_info(self) -> bool: 

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

97 

98 

99class InstanceLogger: 

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

101 

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

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

104 per-instance basis. 

105 

106 The basic functionality is to return a logging level 

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

108 

109 Default implementation is: 

110 

111 'debug' -> logging.DEBUG 

112 True -> logging.INFO 

113 False -> Effective level of underlying logger ( 

114 logging.WARNING by default) 

115 None -> same as False 

116 """ 

117 

118 # Map echo settings to logger levels 

119 _echo_map = { 

120 None: logging.NOTSET, 

121 False: logging.NOTSET, 

122 True: logging.INFO, 

123 "debug": logging.DEBUG, 

124 } 

125 

126 _echo: _EchoFlagType 

127 

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

129 

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

131 self.echo = echo 

132 self.logger = logging.getLogger(name) 

133 

134 # if echo flag is enabled and no handlers, 

135 # add a handler to the list 

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

137 _add_default_handler(self.logger) 

138 

139 # 

140 # Boilerplate convenience methods 

141 # 

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

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

144 

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

146 

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

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

149 

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

151 

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

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

154 

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

156 

157 warn = warning 

158 

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

160 """ 

161 Delegate an error call to the underlying logger. 

162 """ 

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

164 

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

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

167 

168 kwargs["exc_info"] = 1 

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

170 

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

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

173 

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

175 

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

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

178 

179 The level here is determined by the echo 

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

181 logger._log() is called directly. 

182 

183 """ 

184 

185 # inline the logic from isEnabledFor(), 

186 # getEffectiveLevel(), to avoid overhead. 

187 

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

189 return 

190 

191 selected_level = self._echo_map[self.echo] 

192 if selected_level == logging.NOTSET: 

193 selected_level = self.logger.getEffectiveLevel() 

194 

195 if level >= selected_level: 

196 if STACKLEVEL: 

197 kwargs["stacklevel"] = ( 

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

199 ) 

200 

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

202 

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

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

205 

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

207 return False 

208 return level >= self.getEffectiveLevel() 

209 

210 def getEffectiveLevel(self) -> int: 

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

212 

213 level = self._echo_map[self.echo] 

214 if level == logging.NOTSET: 

215 level = self.logger.getEffectiveLevel() 

216 return level 

217 

218 

219def instance_logger( 

220 instance: Identified, echoflag: _EchoFlagType = None 

221) -> None: 

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

223 

224 if instance.logging_name: 

225 name = "%s.%s" % ( 

226 _qual_logger_name_for_cls(instance.__class__), 

227 instance.logging_name, 

228 ) 

229 else: 

230 name = _qual_logger_name_for_cls(instance.__class__) 

231 

232 instance._echo = echoflag # type: ignore[misc] 

233 

234 logger: Union[logging.Logger, InstanceLogger] 

235 

236 if echoflag in (False, None): 

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

238 # avoiding overhead of filtering 

239 logger = logging.getLogger(name) 

240 else: 

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

242 # which checks the flag, overrides normal log 

243 # levels by calling logger._log() 

244 logger = InstanceLogger(echoflag, name) 

245 

246 instance.logger = logger # type: ignore[misc] 

247 

248 

249class echo_property: 

250 __doc__ = """\ 

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

252 

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

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

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

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

257 ``logging.DEBUG``. 

258 """ 

259 

260 @overload 

261 def __get__( 

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

263 ) -> echo_property: ... 

264 

265 @overload 

266 def __get__( 

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

268 ) -> _EchoFlagType: ... 

269 

270 def __get__( 

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

272 ) -> Union[echo_property, _EchoFlagType]: 

273 if instance is None: 

274 return self 

275 else: 

276 return instance._echo 

277 

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

279 instance_logger(instance, echoflag=value)