"""
logbook.compat
~~~~~~~~~~~~~~
Backwards compatibility with stdlib's logging package and the
warnings module.
:copyright: (c) 2010 by Armin Ronacher, Georg Brandl.
:license: BSD, see LICENSE for more details.
"""
import logging
import sys
import warnings
from collections.abc import Mapping
from datetime import date, datetime, timezone
import logbook
[docs]
def redirect_logging(set_root_logger_level=True):
"""Permanently redirects logging to the stdlib. This also
removes all otherwise registered handlers on root logger of
the logging system but leaves the other loggers untouched.
:param set_root_logger_level: controls of the default level of the legacy
root logger is changed so that all legacy log messages get redirected
to Logbook
"""
del logging.root.handlers[:]
logging.root.addHandler(RedirectLoggingHandler())
if set_root_logger_level:
logging.root.setLevel(logging.DEBUG)
[docs]
class redirected_logging:
"""Temporarily redirects logging for all threads and reverts
it later to the old handlers. Mainly used by the internal
unittests::
from logbook.compat import redirected_logging
with redirected_logging():
...
"""
def __init__(self, set_root_logger_level=True):
self.old_handlers = logging.root.handlers[:]
self.old_level = logging.root.level
self.set_root_logger_level = set_root_logger_level
def start(self):
redirect_logging(self.set_root_logger_level)
def end(self, etype=None, evalue=None, tb=None):
logging.root.handlers[:] = self.old_handlers
logging.root.setLevel(self.old_level)
__enter__ = start
__exit__ = end
class LoggingCompatRecord(logbook.LogRecord):
def _format_message(self, msg, *args, **kwargs):
if kwargs:
assert not args
return msg % kwargs
else:
assert not kwargs
return msg % tuple(args)
[docs]
class RedirectLoggingHandler(logging.Handler):
"""A handler for the stdlib's logging system that redirects
transparently to logbook. This is used by the
:func:`redirect_logging` and :func:`redirected_logging`
functions.
If you want to customize the redirecting you can subclass it.
"""
def __init__(self):
logging.Handler.__init__(self)
[docs]
def convert_level(self, level):
"""Converts a logging level into a logbook level."""
if level >= logging.CRITICAL:
return logbook.CRITICAL
if level >= logging.ERROR:
return logbook.ERROR
if level >= logging.WARNING:
return logbook.WARNING
if level >= logging.INFO:
return logbook.INFO
return logbook.DEBUG
[docs]
def find_extra(self, old_record):
"""Tries to find custom data from the old logging record. The
return value is a dictionary that is merged with the log record
extra dictionaries.
"""
rv = vars(old_record).copy()
for key in (
"name",
"msg",
"args",
"levelname",
"levelno",
"pathname",
"filename",
"module",
"exc_info",
"exc_text",
"lineno",
"funcName",
"created",
"msecs",
"relativeCreated",
"thread",
"threadName",
"greenlet",
"processName",
"process",
):
rv.pop(key, None)
return rv
[docs]
def find_caller(self, old_record):
"""Tries to find the caller that issued the call."""
frm = sys._getframe(2)
while frm is not None:
if (
frm.f_globals is globals()
or frm.f_globals is logbook.base.__dict__
or frm.f_globals is logging.__dict__
):
frm = frm.f_back
else:
return frm
[docs]
def convert_time(self, timestamp):
"""Converts the UNIX timestamp of the old record into a
datetime object as used by logbook.
"""
return datetime.utcfromtimestamp(timestamp)
[docs]
def convert_record(self, old_record):
"""Converts an old logging record into a logbook log record."""
args = old_record.args
kwargs = None
# Logging allows passing a mapping object, in which case args will be a mapping.
if isinstance(args, Mapping):
kwargs = args
args = None
record = LoggingCompatRecord(
old_record.name,
self.convert_level(old_record.levelno),
old_record.msg,
args,
kwargs,
old_record.exc_info,
self.find_extra(old_record),
self.find_caller(old_record),
)
record.time = self.convert_time(old_record.created)
return record
[docs]
def emit(self, record):
logbook.dispatch_record(self.convert_record(record))
[docs]
class LoggingHandler(logbook.Handler):
"""Does the opposite of the :class:`RedirectLoggingHandler`, it sends
messages from logbook to logging. Because of that, it's a very bad
idea to configure both.
This handler is for logbook and will pass stuff over to a logger
from the standard library.
Example usage::
from logbook.compat import LoggingHandler, warn
with LoggingHandler():
warn('This goes to logging')
"""
def __init__(self, logger=None, level=logbook.NOTSET, filter=None, bubble=False):
logbook.Handler.__init__(self, level, filter, bubble)
if logger is None:
logger = logging.getLogger()
elif isinstance(logger, str):
logger = logging.getLogger(logger)
self.logger = logger
[docs]
def get_logger(self, record):
"""Returns the logger to use for this record. This implementation
always return :attr:`logger`.
"""
return self.logger
[docs]
def convert_level(self, level):
"""Converts a logbook level into a logging level."""
if level >= logbook.CRITICAL:
return logging.CRITICAL
if level >= logbook.ERROR:
return logging.ERROR
if level >= logbook.WARNING:
return logging.WARNING
if level >= logbook.INFO:
return logging.INFO
return logging.DEBUG
[docs]
def convert_time(self, dt):
"""Converts a datetime object into a timestamp."""
if dt.tzinfo is None:
# Logbook uses naive datetimes to represent UTC (utcnow)
return dt.replace(tzinfo=timezone.utc).timestamp()
return dt.timestamp()
[docs]
def convert_record(self, old_record):
"""Converts a record from logbook to logging."""
record = logging.LogRecord(
old_record.channel,
self.convert_level(old_record.level),
old_record.filename,
old_record.lineno,
old_record.message,
(),
old_record.exc_info,
func=old_record.func_name,
)
for key, value in old_record.extra.items():
record.__dict__.setdefault(key, value)
record.created = self.convert_time(old_record.time)
return record
[docs]
def emit(self, record):
self.get_logger(record).handle(self.convert_record(record))
[docs]
def redirect_warnings():
"""Like :func:`redirected_warnings` but will redirect all warnings
to the shutdown of the interpreter:
.. code-block:: python
from logbook.compat import redirect_warnings
redirect_warnings()
"""
redirected_warnings().__enter__()
[docs]
class redirected_warnings:
"""A context manager that copies and restores the warnings filter upon
exiting the context, and logs warnings using the logbook system.
The :attr:`~logbook.LogRecord.channel` attribute of the log record will be
the import name of the warning.
Example usage:
.. code-block:: python
from logbook.compat import redirected_warnings
from warnings import warn
with redirected_warnings():
warn(DeprecationWarning('logging should be deprecated'))
"""
def __init__(self):
self._entered = False
def message_to_unicode(self, message):
return str(message)
def make_record(self, message, exception, filename, lineno):
category = exception.__name__
if exception.__module__ not in ("exceptions", "builtins"):
category = exception.__module__ + "." + category
rv = logbook.LogRecord(category, logbook.WARNING, message)
# we don't know the caller, but we get that information from the
# warning system. Just attach them.
rv.filename = filename
rv.lineno = lineno
return rv
def start(self):
if self._entered: # pragma: no cover
raise RuntimeError("Cannot enter %r twice" % self)
self._entered = True
self._filters = warnings.filters
warnings.filters = self._filters[:]
self._showwarning = warnings.showwarning
def showwarning(message, category, filename, lineno, file=None, line=None):
message = self.message_to_unicode(message)
record = self.make_record(message, category, filename, lineno)
logbook.dispatch_record(record)
warnings.showwarning = showwarning
def end(self, etype=None, evalue=None, tb=None):
if not self._entered: # pragma: no cover
raise RuntimeError("Cannot exit %r without entering first" % self)
warnings.filters = self._filters
warnings.showwarning = self._showwarning
__enter__ = start
__exit__ = end
Logbook