Source code for firebird.base.logging

# SPDX-FileCopyrightText: 2020-present The Firebird Projects <www.firebirdsql.org>
#
# SPDX-License-Identifier: MIT
#
# PROGRAM/MODULE: firebird-base
# FILE:           firebird/base/logging.py
# DESCRIPTION:    Context-based logging
# CREATED:        14.5.2020
#
# The contents of this file are subject to the MIT License
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#
# Copyright (c) 2020 Firebird Project (www.firebirdsql.org)
# All Rights Reserved.
#
# Contributor(s): Pavel Císař (original code)
#                 ______________________________________

"""firebird-base - Context-based logging

"""

from __future__ import annotations
from typing import Any, Dict, Tuple, Union, Hashable
from enum import IntEnum, Flag, auto
from collections.abc import Mapping
from dataclasses import dataclass
from logging import Logger, LoggerAdapter, getLogger, lastResort, Formatter
from platform import python_version_tuple
from .types import UNDEFINED, DEFAULT, ANY, ALL, Distinct, CachedDistinct, Sentinel
from .collections import Registry

[docs] class LogLevel(IntEnum): """Shadow enumeration for logging levels. """ NOTSET = 0 DEBUG = 10 INFO = 20 WARNING = 30 ERROR = 40 CRITICAL = 50 FATAL = CRITICAL WARN = WARNING
[docs] class BindFlag(Flag): """Internal flags used by `LoggingManager`. """ DIRECT = auto() ANY_AGENT = auto() ANY_CTX = auto() ANY_ANY = auto()
[docs] class FBLoggerAdapter(LoggerAdapter, CachedDistinct): """`~logging.LoggerAdapter` that injects information about context, agent and topic into `extra` and with **f-string** log message support. """
[docs] def __init__(self, logger: Logger, agent: Any=UNDEFINED, context: Any=UNDEFINED, topic: str=''): """ Arguments: logger: Adapted Logger instance. agent: Agent for logger context: Context for logger topic: Topic of recorded information. """ #: Adapted Logger instance. self.logger: Logger = logger #: Agent for logger. self.agent: Any = agent #: Context for logger. self.context: Any = context #: Topic for logger. self.topic: str = topic
[docs] @classmethod def extract_key(cls, *args, **kwargs) -> Hashable: """Returns instance key extracted from constructor arguments. """ return (args[1], args[2])
[docs] def get_key(self) -> Hashable: # pragma: no cover """Returns instance key. """ return (self.topic, self.agent, self.context)
[docs] def process(self, msg, kwargs) -> Tuple[str, Dict]: """Process the logging message and keyword arguments passed into a logging call to insert contextual information. You can either manipulate the message itself, the keyword args or both. Return the message and kwargs modified (or not) to suit your needs. """ return msg, kwargs
[docs] def log(self, level, msg, *args, **kwargs): """Delegate a log call to the underlying logger after processing. Interpolates the message as **f-string** using either `kwargs` or dict passed as only one positional argument. If sole positional argument is not dictionary or `args` has more than one item, adds `args` into namespace for interpolation. Moves 'context', 'agent' and 'topic' keyword arguments into `extra`. Strips out all keyword arguments not expected by `logging.Logger`. """ if self.isEnabledFor(level): msg, kwargs = self.process(msg, kwargs) if (args and len(args) == 1 and isinstance(args[0], Mapping) and args[0]): ns = args[0] else: ns = kwargs if args: ns['args'] = args msg = eval(f'f"""{msg}"""', globals(), ns) args = () if 'stacklevel' not in kwargs: kwargs['stacklevel'] = 3 if int(python_version_tuple()[1]) < 11 else 2 kwargs.setdefault('extra', {}).update(topic=self.topic, agent=self.agent, context=self.context) self.logger.log(level, msg, *args, **{k: v for k, v in kwargs.items() if k in ['exc_info', 'stack_info', 'stacklevel', 'extra']})
@dataclass(order=True, frozen=True) class BindInfo(Distinct): """Information about Logger binding. """ topic: str agent: str context: str logger: FBLoggerAdapter def get_key(self) -> Any: "Returns distinct key value = Tuple(topic, agent, context)." return (self.topic, self.agent, self.context)
[docs] def get_logging_id(obj: Any) -> Any: """Returns logging ID for object. Arguments: obj: Any object Returns: 1. `logging_id` attribute if `obj` does have it, or.. 2. `__qualname__` attribute if `obj` does have it, or.. 3. `str(obj)` """ return getattr(obj, 'logging_id', getattr(obj, '__qualname__', str(obj)))
[docs] class LoggingIdMixin: """Mixin class that adds `logging_id` property and `__str__` that returns `logging_id`. """ def __str__(self): return self.logging_id @property def logging_id(self) -> str: """Returns `_logging_id_` attribute if defined, else returns qualified class name. """ return getattr(self, '_logging_id_', self.__class__.__qualname__)
[docs] class LoggingManager: """Logger manager. """ def __init__(self): self.loggers: Registry = Registry() self.topics: Dict[str, int] = {} self.bindings: BindFlag = BindFlag(0) def _update_bindings(self, agent: Any, context: Any) -> None: if agent is ANY: self.bindings |= BindFlag.ANY_AGENT if context is ANY: self.bindings |= BindFlag.ANY_CTX if (agent is ANY) and (context is ANY): self.bindings |= BindFlag.ANY_ANY if (agent is not ANY) and (context is not ANY): self.bindings |= BindFlag.DIRECT def _update_topics(self, topic: str) -> None: if topic in self.topics: self.topics[topic] += 1 else: self.topics[topic] = 1
[docs] def bind_logger(self, agent: Any, context: Any, logger: Union[str, Logger], topic: str='') -> None: """Bind agent and context to specific logger. Arguments: agent: Agent identification context: Context identification logger: Loger (instance or name) topic: Topic of recorded information The identification of agent and context could be: 1. String 2. Object instance. Uses `get_logging_id()` to retrieve its logging ID. 3. Sentinel. The ANY sentinel matches any particular agent or context. You can use sentinel `.UNDEFINED` to register a logger for cases when agent or context are not specified in logger lookup. Important: You SHOULD NOT use sentinel `.ALL` for `agent` or `context` identification! This sentinel is used by `.unbind()`, so bindings that use ALL could not be removed by `.unbind()`. """ if isinstance(logger, str): logger = getLogger(logger) if not isinstance(agent, (str, Sentinel)): agent = get_logging_id(agent) if not isinstance(context, (str, Sentinel)): context = get_logging_id(context) if agent is not ANY and context is not ANY: logger = FBLoggerAdapter(logger, agent, context) self._update_bindings(agent, context) self._update_topics(topic) self.loggers.update(BindInfo(topic, agent, context, logger))
[docs] def unbind(self, agent: Any, context: Any, topic: str='') -> int: """Drops logger bindings. """ if not isinstance(agent, (str, Sentinel)): agent = get_logging_id(agent) if not isinstance(context, (str, Sentinel)): context = get_logging_id(context) if topic in self.topics: rm = [i for i in self.loggers if i.topic == topic and ((i.agent == agent) or agent is ALL) and ((i.context == context) or context is ALL)] for item in rm: self.loggers.remove(item) # recalculate optimizations self.topics.clear() self.bindings = BindFlag(0) for item in self.loggers: self._update_bindings(item.agent, item.context) self._update_topics(item.topic) return len(rm) return 0
[docs] def clear(self) -> None: """Remove all logger bindings. """ self.loggers.clear() self.topics.clear() self.bindings = BindFlag(0)
[docs] def get_logger(self, agent: Any=UNDEFINED, context: Any=DEFAULT, topic: str='') -> FBLoggerAdapter: """Return a logger for the specified agent and context combination. Arguments: agent: Agent identification. context: Context identification. topic: Topic of recorded information. The identification of agent and context could be: 1. String 2. Object instance. Uses `get_logging_id()` to retrieve its logging ID. 3. Sentinel `.UNDEFINED` 4. When `context` is sentinel `.DEFAULT`, uses `agent` attribute `log_context` (if defined) or sentinel `.UNDEFINED` otherwise. The search for a suitable topic logger proceeds as follows: 1. Return logger registered for specified agent and context, or... 2. Return logger registered for ANY agent and specified context, or... 3. Return logger registered for specified agent and ANY context, or... 4. Return logger registered for ANY agent and ANY context, or... 5. Return the root logger. """ if context is DEFAULT: context = getattr(agent, 'log_context', UNDEFINED) if agent is not UNDEFINED and not isinstance(agent, str): agent = get_logging_id(agent) if context is not UNDEFINED and not isinstance(context, str): context = get_logging_id(context) result: BindInfo = None if topic in self.topics: if BindFlag.DIRECT in self.bindings and \ (result := self.loggers.get((topic, agent, context))) is not None: result = result.logger elif BindFlag.ANY_AGENT in self.bindings and \ (result := self.loggers.get((topic, ANY, context))) is not None: result = result.logger elif BindFlag.ANY_CTX in self.bindings and \ (result := self.loggers.get((topic, agent, ANY))) is not None: result = result.logger elif BindFlag.ANY_ANY in self.bindings and \ (result := self.loggers.get((topic, ANY, ANY))) is not None: result = result.logger else: result = getLogger(topic) else: result = getLogger(topic) return result if isinstance(result, FBLoggerAdapter) \ else FBLoggerAdapter(result, agent, context, topic)
#: Logging Manager logging_manager: LoggingManager = LoggingManager() #: shortcut for `logging_manager.bind_logger()` bind_logger = logging_manager.bind_logger #: shortcut for `logging_manager.get_logger()` get_logger = logging_manager.get_logger # Install simple formatter for lastResort handler if lastResort is not None and lastResort.formatter is None: lastResort.setFormatter(Formatter('%(levelname)s: %(message)s'))
[docs] def install_null_logger(): """Installs 'null' logger. """ log = getLogger('null') log.propagate = False log.disabled = True