"""
Cement logging extension module.
"""
import os
import logging
from ..core import log
from ..core.deprecations import deprecate
from ..utils.misc import is_true, minimal_logger
from ..utils import fs
LOG = minimal_logger(__name__)
try: # pragma: no cover
NullHandler = logging.NullHandler # pragma: no cover
except AttributeError: # pragma: no cover
# Not supported on Python < 3.1/2.7 # pragma: no cover
class NullHandler(logging.Handler): # pragma: no cover
def handle(self, record): # pragma: no cover
pass # pragma: no cover
# pragma: no cover
def emit(self, record): # pragma: no cover
pass # pragma: no cover
# pragma: no cover
def createLock(self): # pragma: no cover
self.lock = None # pragma: no cover
[docs]
class LoggingLogHandler(log.LogHandler):
"""
This class is an implementation of the :ref:`Log <cement.core.log>`
interface, and sets up the logging facility using the standard Python
`logging <http://docs.python.org/library/logging.html>`_ module.
"""
levels = ['INFO', 'WARNING', 'ERROR', 'DEBUG', 'FATAL', 'CRITICAL']
def __init__(self, *args, **kw):
super(LoggingLogHandler, self).__init__(*args, **kw)
self.app = None
[docs]
def _setup(self, app_obj):
super(LoggingLogHandler, self)._setup(app_obj)
if self._meta.namespace is None:
self._meta.namespace = "%s" % self.app._meta.label
self.backend = logging.getLogger("cement:app:%s" %
self._meta.namespace)
# hack for application debugging
if is_true(self.app._meta.debug):
self.app.config.set(self._meta.config_section, 'level', 'DEBUG')
level = self.app.config.get(self._meta.config_section, 'level')
self.set_level(level)
self.backend.propagate = self._meta.propagate
LOG.debug("logging initialized for '%s' using %s" %
(self._meta.namespace, self.__class__.__name__))
[docs]
def set_level(self, level):
"""
Set the log level. Must be one of the log levels configured in
self.levels which are
``['INFO', 'WARNING', 'ERROR', 'DEBUG', 'FATAL', 'CRITICAL]``.
As of Cement 3.0.10, the FATAL facility is deprecated and will be
removed in future versions of Cement. Please us `CRITICAL` instead.
:param level: The log level to set.
"""
self.clear_loggers(self._meta.namespace)
for namespace in self._meta.clear_loggers:
self.clear_loggers(namespace)
level = level.upper()
if level not in self.levels:
level = 'INFO'
if level == 'FATAL':
deprecate('3.0.10-1')
level = getattr(logging, level.upper())
self.backend.setLevel(level)
# console
self._setup_console_log()
# file
self._setup_file_log()
[docs]
def get_level(self):
"""Returns the current log level."""
return logging.getLevelName(self.backend.level)
[docs]
def clear_loggers(self, namespace):
"""Clear any previously configured loggers for ``namespace``."""
for i in logging.getLogger("cement:app:%s" % namespace).handlers:
logging.getLogger("cement:app:%s" % namespace).removeHandler(i)
self.backend = logging.getLogger("cement:app:%s" % namespace)
def _get_console_format(self):
if self.get_level() == logging.getLevelName(logging.DEBUG):
format = self._meta.debug_format
else:
format = self._meta.console_format
return format
def _get_file_format(self):
if self.get_level() == logging.getLevelName(logging.DEBUG):
format = self._meta.debug_format
else:
format = self._meta.file_format
return format
def _get_file_formatter(self, format):
return self._meta.formatter_class(format)
def _get_console_formatter(self, format):
return self._meta.formatter_class(format)
[docs]
def _setup_console_log(self):
"""Add a console log handler."""
namespace = self._meta.namespace
to_console = self.app.config.get(self._meta.config_section,
'to_console')
if is_true(to_console):
console_handler = logging.StreamHandler()
format = self._get_console_format()
formatter = self._get_console_formatter(format)
console_handler.setFormatter(formatter)
console_handler.setLevel(getattr(logging, self.get_level()))
else:
console_handler = NullHandler()
# FIXME: self._clear_loggers() should be preventing this but its not!
for i in logging.getLogger("cement:app:%s" % namespace).handlers:
if isinstance(i, logging.StreamHandler):
self.backend.removeHandler(i)
self.backend.addHandler(console_handler)
[docs]
def _setup_file_log(self):
"""Add a file log handler."""
namespace = self._meta.namespace
file_path = self.app.config.get(self._meta.config_section, 'file')
rotate = self.app.config.get(self._meta.config_section, 'rotate')
max_bytes = self.app.config.get(self._meta.config_section,
'max_bytes')
max_files = self.app.config.get(self._meta.config_section,
'max_files')
if file_path:
file_path = fs.abspath(file_path)
log_dir = os.path.dirname(file_path)
if not os.path.exists(log_dir):
os.makedirs(log_dir)
if rotate:
from logging.handlers import RotatingFileHandler
file_handler = RotatingFileHandler(
file_path,
maxBytes=int(max_bytes),
backupCount=int(max_files),
)
else:
from logging import FileHandler
file_handler = FileHandler(file_path)
format = self._get_file_format()
formatter = self._get_file_formatter(format)
file_handler.setFormatter(formatter)
file_handler.setLevel(getattr(logging, self.get_level()))
else:
file_handler = NullHandler()
# FIXME: self._clear_loggers() should be preventing this but its not!
for i in logging.getLogger("cement:app:%s" % namespace).handlers:
if isinstance(i, file_handler.__class__): # pragma: nocover
self.backend.removeHandler(i) # pragma: nocover
self.backend.addHandler(file_handler)
def _get_logging_kwargs(self, namespace, **kw):
if namespace is None:
namespace = self._meta.namespace
if 'extra' in kw.keys() and 'namespace' in kw['extra'].keys():
pass
elif 'extra' in kw.keys() and 'namespace' not in kw['extra'].keys():
kw['extra']['namespace'] = namespace
else:
kw['extra'] = dict(namespace=namespace)
return kw
[docs]
def info(self, msg, namespace=None, **kw):
"""
Log to the INFO facility.
Args:
msg (str): The message to log.
Keyword Args:
namespace (str): A log prefix, generally the module ``__name__``
that the log is coming from. Will default to
``self._meta.namespace`` if none is passed.
Other Parameters:
kwargs: Keyword arguments are passed on to the backend logging
system.
"""
kwargs = self._get_logging_kwargs(namespace, **kw)
self.backend.info(msg, **kwargs)
[docs]
def warning(self, msg, namespace=None, **kw):
"""
Log to the WARNING facility.
Args:
msg (str): The message to log.
Keyword Args:
namespace (str): A log prefix, generally the module ``__name__``
that the log is coming from. Will default to
``self._meta.namespace`` if none is passed.
Other Parameters:
kwargs: Keyword arguments are passed on to the backend logging
system.
"""
kwargs = self._get_logging_kwargs(namespace, **kw)
self.backend.warning(msg, **kwargs)
[docs]
def error(self, msg, namespace=None, **kw):
"""
Log to the ERROR facility.
:Args:
msg (str): The message to log.
Keyword Args:
namespace (str): A log prefix, generally the module ``__name__``
that the log is coming from. Will default to
``self._meta.namespace`` if none is passed.
Other Parameters:
kwargs: Keyword arguments are passed on to the backend logging
system.
"""
kwargs = self._get_logging_kwargs(namespace, **kw)
self.backend.error(msg, **kwargs)
[docs]
def critical(self, msg, namespace=None, **kw):
"""
Log to the CRITICAL facility.
Args:
msg (str): The message to log.
Keyword Args:
namespace (str): A log prefix, generally the module ``__name__``
that the log is coming from. Will default to
``self._meta.namespace`` if none is passed.
Other Parameters:
kwargs: Keyword arguments are passed on to the backend logging
system.
"""
kwargs = self._get_logging_kwargs(namespace, **kw)
self.backend.critical(msg, **kwargs)
[docs]
def fatal(self, msg, namespace=None, **kw):
"""
Log to the FATAL (aka CRITICAL) facility.
As of Cement 3.0.10, this method is deprecated and will be removed in
future versions of Cement. Please us `critical()` instead.
Args:
msg (str): The message to log.
Keyword Args:
namespace (str): A log prefix, generally the module ``__name__``
that the log is coming from. Will default to
``self._meta.namespace`` if none is passed.
Other Parameters:
kwargs: Keyword arguments are passed on to the backend logging
system.
"""
deprecate('3.0.10-1')
kwargs = self._get_logging_kwargs(namespace, **kw)
self.backend.fatal(msg, **kwargs)
[docs]
def debug(self, msg, namespace=None, **kw):
"""
Log to the DEBUG facility.
Args:
msg (str): The message to log.
Keyword Args:
namespace (str): A log prefix, generally the module ``__name__``
that the log is coming from. Will default to
``self._meta.namespace`` if none is passed.
Other Parameters:
kwargs: Keyword arguments are passed on to the backend logging
system.
"""
kwargs = self._get_logging_kwargs(namespace, **kw)
self.backend.debug(msg, **kwargs)
def add_logging_arguments(app):
if app.log._meta.log_level_argument is not None:
app.args.add_argument(*app.log._meta.log_level_argument,
dest='log_logging_level',
help=app.log._meta.log_level_argument_help,
choices=[x.lower() for x in app.log.levels])
def handle_logging_arguments(app):
if hasattr(app.pargs, 'log_logging_level'):
if app.pargs.log_logging_level is not None:
app.log.set_level(app.pargs.log_logging_level)
if app.pargs.log_logging_level in ['debug', 'DEBUG']:
app._meta.debug = True
def load(app):
app.handler.register(LoggingLogHandler)
app.hook.register('pre_argument_parsing', add_logging_arguments)
app.hook.register('post_argument_parsing', handle_logging_arguments)