# coding: utf-8
"""
Law logging setup.
"""
__all__ = [
"get_logger", "setup_logging", "setup_logger", "create_stream_handler", "is_tty_handler",
"get_tty_handlers", "Logger", "LogFormatter",
]
from collections import defaultdict
import logging
import six
from law.util import no_value, colored, ipykernel, ON_COLAB
_logging_setup = False
[docs]def setup_logging():
"""
Sets up the internal law loggers as well as all other loggers listed in the ``"logging"`` config
section as (*name*, *level*) pairs. This includes loggers that do not use the ``"law.*"``
namespace which can be seen as a convenient feature to set up custom loggers.
"""
global _logging_setup
# make sure logging is setup only once
if _logging_setup:
return
_logging_setup = True
# remove root handlers on colab
if ON_COLAB and logging.root.handlers:
logging.root.removeHandler(logging.root.handlers[0])
# setup the main law logger first and set its handler which is propagated to subloggers
logger = get_logger("law", skip_setup=True)
logger = setup_logger(logger, add_console_handler=False)
logger.addHandler(create_stream_handler())
# set levels for all loggers and add the console handler for all non-law loggers
from law.config import Config
for name, level in Config.instance().items("logging"):
setup_logger(name, level)
def _logger_setup(logger, value=None):
attr = "_law_logger_setup"
if value is not None:
setattr(logger, attr, value)
return getattr(logger, attr, False)
[docs]def get_logger(*args, **kwargs):
""" get_logger(*args, skip_setup=False, **kwargs)
Replacement for *logging.getLogger* that makes sure that the custom :py:class:`Logger` class is
used when new loggers are created and that the logger is properly set up by
:py:meth:`setup_logger`.
"""
skip_setup = kwargs.pop("skip_setup", False)
orig_cls = logging.getLoggerClass()
logging.setLoggerClass(Logger)
try:
logger = logging.getLogger(*args, **kwargs)
# set it up once
if not skip_setup:
setup_logger(logger)
return logger
finally:
logging.setLoggerClass(orig_cls)
[docs]def setup_logger(logger, level=None, add_console_handler=None, clear=False, force=False):
"""
Sets up a *logger*, optionally given by its name, configures it to have a certain *level* and
adds a preconfigured console handler when *add_console_handler* is *True*. When
*add_console_handler* is a dictionary, its items are forwarded as keyword arguments to the
:py:func:`create_stream_handler` which handles the handler setup internally. When *None*,
*add_console_handler* is default to *True* in case the logger is not a "law" sublogger and has
no tty handlers registered yet.
Each logger is setup only once unless *force* is *True.
*level* can either be an integer or the name of a level present in the *logging* module. When no
*level* is given, the level of the ``"law"`` base logger is used as a default. When the logger
already existed and *clear* is *True*, all handlers and filters are removed first. The logger
object is returned.
"""
# get the logger
logger = logger if isinstance(logger, logging.Logger) else get_logger(logger, skip_setup=True)
name = logger.name
# do nothing when the logger was already set up or force is defined
if _logger_setup(logger) or force:
return logger
_logger_setup(logger, True)
# sanitize the level
if level is None:
from law.config import Config
level = Config.instance().get_expanded("logging", name, None)
if isinstance(level, six.string_types):
level = getattr(logging, level.upper(), None)
if level is None:
level = get_logger("law").level
# clear handlers and filters
is_existing = name in logging.root.manager.loggerDict
if is_existing and clear:
for h in list(logger.handlers):
logger.removeHandler(h)
for f in list(logger.filters):
logger.removeFilter(f)
# set the level
logger.setLevel(level)
# add a console handler
if add_console_handler is None:
add_console_handler = not name.startswith("law.") and not get_tty_handlers(name)
if add_console_handler or isinstance(add_console_handler, dict):
kwargs = add_console_handler if isinstance(add_console_handler, dict) else {}
logger.addHandler(create_stream_handler(**kwargs))
return logger
[docs]def create_stream_handler(handler_kwargs=None, formatter_kwargs=None, formatter_cls=no_value):
""" create_stream_handler(handler_kwargs=None, formatter_kwargs=None, formatter_cls=LogFormatter)
Creates a new StreamHandler instance, passing all *handler_kwargs* to its constructor, and
returns it. When not *None*, an instance of *formatter_cls* is created using *formatter_kwargs*
and added to the handler instance.
"""
# create the handler
handler = logging.StreamHandler(**(handler_kwargs or {}))
# add a formatter
if formatter_cls == no_value:
formatter_cls = LogFormatter
if formatter_cls is not None:
formatter = formatter_cls(**(formatter_kwargs or {}))
handler.setFormatter(formatter)
return handler
[docs]def is_tty_handler(handler):
"""
Returns *True* if a logging *handler* is a *StreamHandler* which logs to a tty (i.e. *stdout* or
*stderr*), an IPython *OutStream*, or a base *Handler* with a *console* attribute evaluating to
*True*. The latter check is intended to cover a variety of handlers provided by custom modules.
"""
if isinstance(handler, logging.StreamHandler) and getattr(handler, "stream", None):
if callable(getattr(handler.stream, "isatty", None)) and handler.stream.isatty():
return True
elif ipykernel and isinstance(handler.stream, ipykernel.iostream.OutStream):
return True
if isinstance(handler, logging.Handler) and getattr(handler, "console", None):
return True
return False
[docs]def get_tty_handlers(logger):
"""
Returns a list of all handlers of a *logger* that log to a tty.
"""
if isinstance(logger, six.string_types):
logger = get_logger(logger)
return [handler for handler in getattr(logger, "handlers", []) if is_tty_handler(handler)]
[docs]class Logger(logging.Logger):
"""
Custom logger class that adds an additional set of log methods, i.e., :py:meth:`debug_once`,
:py:meth:`info_once`, :py:meth:`warning_once`, :py:meth:`error_once`, :py:meth:`critical_once`
and :py:meth:`fatal_once`, that log certain messages only once depending on a string identifier.
"""
def __init__(self, *args, **kwargs):
super(Logger, self).__init__(*args, **kwargs)
# names of logs per level that are issued only once
self._once_logs = defaultdict(set)
def debug_once(self, log_id, *args, **kwargs):
# when no log_id is set, but just a message, it is received as log_id
if not args:
args = (log_id,)
if log_id not in self._once_logs["debug"]:
self._once_logs["debug"].add(log_id)
self.debug(*args, **kwargs)
def info_once(self, log_id, *args, **kwargs):
# when no log_id is set, but just a message, it is received as log_id
if not args:
args = (log_id,)
if log_id not in self._once_logs["info"]:
self._once_logs["info"].add(log_id)
self.info(*args, **kwargs)
def warning_once(self, log_id, *args, **kwargs):
# when no log_id is set, but just a message, it is received as log_id
if not args:
args = (log_id,)
if log_id not in self._once_logs["warning"]:
self._once_logs["warning"].add(log_id)
self.warning(*args, **kwargs)
def error_once(self, log_id, *args, **kwargs):
# when no log_id is set, but just a message, it is received as log_id
if not args:
args = (log_id,)
if log_id not in self._once_logs["error"]:
self._once_logs["error"].add(log_id)
self.error(*args, **kwargs)
def critical_once(self, log_id, *args, **kwargs):
# when no log_id is set, but just a message, it is received as log_id
if not args:
args = (log_id,)
if log_id not in self._once_logs["critical"]:
self._once_logs["critical"].add(log_id)
self.critical(*args, **kwargs)
def fatal_once(self, log_id, *args, **kwargs):
# when no log_id is set, but just a message, it is received as log_id
if not args:
args = (log_id,)
if log_id not in self._once_logs["fatal"]:
self._once_logs["fatal"].add(log_id)
self.fatal(*args, **kwargs)
