codekingpro/portable-devtools

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"""
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)
289