logging.py 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145
  1. import contextlib
  2. from functools import (
  3. cached_property,
  4. )
  5. import logging
  6. from typing import (
  7. Any,
  8. Dict,
  9. Iterator,
  10. Tuple,
  11. Type,
  12. TypeVar,
  13. Union,
  14. cast,
  15. )
  16. from .toolz import (
  17. assoc,
  18. )
  19. DEBUG2_LEVEL_NUM = 8
  20. TLogger = TypeVar("TLogger", bound=logging.Logger)
  21. class ExtendedDebugLogger(logging.Logger):
  22. """
  23. Logging class that can be used for lower level debug logging.
  24. """
  25. @cached_property
  26. def show_debug2(self) -> bool:
  27. return self.isEnabledFor(DEBUG2_LEVEL_NUM)
  28. def debug2(self, message: str, *args: Any, **kwargs: Any) -> None:
  29. if self.show_debug2:
  30. self.log(DEBUG2_LEVEL_NUM, message, *args, **kwargs)
  31. else:
  32. # When we find that `DEBUG2` isn't enabled we completely replace
  33. # the `debug2` function in this instance of the logger with a noop
  34. # lambda to further speed up
  35. self.__dict__["debug2"] = lambda message, *args, **kwargs: None
  36. def __reduce__(self) -> Tuple[Any, ...]:
  37. # This is needed because our parent's implementation could
  38. # cause us to become a regular Logger on unpickling.
  39. return get_extended_debug_logger, (self.name,)
  40. def setup_DEBUG2_logging() -> None:
  41. """
  42. Installs the `DEBUG2` level logging levels to the main logging module.
  43. """
  44. if not hasattr(logging, "DEBUG2"):
  45. logging.addLevelName(DEBUG2_LEVEL_NUM, "DEBUG2")
  46. logging.DEBUG2 = DEBUG2_LEVEL_NUM # type: ignore
  47. @contextlib.contextmanager
  48. def _use_logger_class(logger_class: Type[logging.Logger]) -> Iterator[None]:
  49. original_logger_class = logging.getLoggerClass()
  50. logging.setLoggerClass(logger_class)
  51. try:
  52. yield
  53. finally:
  54. logging.setLoggerClass(original_logger_class)
  55. def get_logger(name: str, logger_class: Union[Type[TLogger], None] = None) -> TLogger:
  56. if logger_class is None:
  57. return cast(TLogger, logging.getLogger(name))
  58. else:
  59. with _use_logger_class(logger_class):
  60. # The logging module caches logger instances. The following code
  61. # ensures that if there is a cached instance that we don't
  62. # accidentally return the incorrect logger type because the logging
  63. # module does not *update* the cached instance in the event that
  64. # the global logging class changes.
  65. #
  66. # types ignored b/c mypy doesn't identify presence of
  67. # manager on logging.Logger
  68. manager = logging.Logger.manager
  69. if name in manager.loggerDict:
  70. if type(manager.loggerDict[name]) is not logger_class:
  71. del manager.loggerDict[name]
  72. return cast(TLogger, logging.getLogger(name))
  73. def get_extended_debug_logger(name: str) -> ExtendedDebugLogger:
  74. return get_logger(name, ExtendedDebugLogger)
  75. THasLoggerMeta = TypeVar("THasLoggerMeta", bound="HasLoggerMeta")
  76. class HasLoggerMeta(type):
  77. """
  78. Assigns a logger instance to a class, derived from the import path and name.
  79. This metaclass uses `__qualname__` to identify a unique and meaningful name
  80. to use when creating the associated logger for a given class.
  81. """
  82. logger_class = logging.Logger
  83. def __new__(
  84. mcls: Type[THasLoggerMeta],
  85. name: str,
  86. bases: Tuple[Type[Any]],
  87. namespace: Dict[str, Any],
  88. ) -> THasLoggerMeta:
  89. if "logger" in namespace:
  90. # If a logger was explicitly declared we shouldn't do anything to
  91. # replace it.
  92. return super().__new__(mcls, name, bases, namespace)
  93. if "__qualname__" not in namespace:
  94. raise AttributeError("Missing __qualname__")
  95. with _use_logger_class(mcls.logger_class):
  96. logger = logging.getLogger(namespace["__qualname__"])
  97. return super().__new__(mcls, name, bases, assoc(namespace, "logger", logger))
  98. @classmethod
  99. def replace_logger_class(
  100. mcls: Type[THasLoggerMeta], value: Type[logging.Logger]
  101. ) -> Type[THasLoggerMeta]:
  102. return type(mcls.__name__, (mcls,), {"logger_class": value})
  103. @classmethod
  104. def meta_compat(
  105. mcls: Type[THasLoggerMeta], other: Type[type]
  106. ) -> Type[THasLoggerMeta]:
  107. return type(mcls.__name__, (mcls, other), {})
  108. class HasLogger(metaclass=HasLoggerMeta):
  109. logger: logging.Logger
  110. HasExtendedDebugLoggerMeta = HasLoggerMeta.replace_logger_class(ExtendedDebugLogger)
  111. class HasExtendedDebugLogger(metaclass=HasExtendedDebugLoggerMeta): # type: ignore
  112. logger: ExtendedDebugLogger