Source code for pyscaffold.log
# -*- coding: utf-8 -*-
"""
Custom logging infrastructure to provide execution information for the user.
"""
from __future__ import absolute_import, print_function
from collections import defaultdict
from contextlib import contextmanager
from logging import INFO, Formatter, LoggerAdapter, StreamHandler, getLogger
from os.path import realpath, relpath
from . import termui
DEFAULT_LOGGER = __name__
def _are_equal_paths(path1, path2):
return realpath(path1) == realpath(path2)
def _is_current_path(path):
return _are_equal_paths(path, '.')
[docs]def configure_logger(opts):
"""Configure the default logger
Args:
opts (dict): command line parameters
"""
if 'log_level' in opts:
getLogger(DEFAULT_LOGGER).setLevel(opts['log_level'])
# if terminal supports, use colors
stream = getattr(logger.handler, 'stream', None)
if termui.supports_color(stream):
logger.formatter = ColoredReportFormatter()
logger.handler.setFormatter(logger.formatter)
[docs]class ReportFormatter(Formatter):
"""Formatter that understands custom fields in the log record."""
ACTIVITY_MAXLEN = 12
SPACING = ' '
CONTEXT_PREFIX = 'from'
TARGET_PREFIX = 'to'
[docs] def format(self, record):
"""Compose message when a record with report information is given."""
if hasattr(record, 'activity'):
return self.format_report(record)
return self.format_default(record)
[docs] def create_padding(self, activity):
"""Create the appropriate padding in order to align activities."""
actual = len(activity)
count = max(self.ACTIVITY_MAXLEN - actual, 0)
return ' ' * count
[docs] def format_path(self, path):
"""Simplify paths to avoid wasting space in terminal."""
if path[0] in './~':
# Heuristic to determine if subject is a file path
# that needs to be made short
abbrev = relpath(path)
if len(abbrev) < len(path):
# Ignore if not shorter
path = abbrev
return path
# Subclasses may need the activity name to choose correct format,
# so the following 3 methods accept a second parameter
# (even if they not use it)
[docs] def format_subject(self, subject, _activity=None):
"""Format the subject of the activity."""
return self.format_path(subject)
[docs] def format_target(self, target, _activity=None):
"""Format extra information about the activity target."""
if target and not _is_current_path(target):
return self.TARGET_PREFIX + ' ' + repr(self.format_path(target))
return ''
[docs] def format_context(self, context, _activity=None):
"""Format extra information about the activity context."""
if context and not _is_current_path(context):
return self.CONTEXT_PREFIX + ' ' + repr(self.format_path(context))
return ''
[docs] def format_default(self, record):
"""Format default log messages."""
record.msg = self.SPACING * max(record.nesting, 0) + record.msg
return super(ReportFormatter, self).format(record)
[docs] def format_report(self, record):
"""Compose message when a custom record is given."""
activity = record.activity
record.msg = (
self.create_padding(activity) +
self.format_activity(activity) +
self.SPACING * max(record.nesting + 1, 0) +
' '.join([
text for text in [
self.format_subject(record.subject, activity),
self.format_target(record.target, activity),
self.format_context(record.context, activity)
] if text # Filter empty strings
])
)
return super(ReportFormatter, self).format(record)
[docs]class ColoredReportFormatter(ReportFormatter):
"""Format logs with ANSI colors."""
ACTIVITY_STYLES = defaultdict(
lambda: ('blue', 'bold'),
create=('green', 'bold'),
move=('green', 'bold'),
remove=('red', 'bold'),
delete=('red', 'bold'),
skip=('yellow', 'bold'),
run=('magenta', 'bold'),
invoke=('bold',)
)
SUBJECT_STYLES = defaultdict(
tuple,
invoke=('blue',)
)
LOG_STYLES = defaultdict(
tuple,
debug=('green',),
info=('blue',),
warning=('yellow',),
error=('red',),
critical=('red', 'bold')
)
CONTEXT_PREFIX = termui.decorate(
ReportFormatter.CONTEXT_PREFIX, 'magenta', 'bold')
TARGET_PREFIX = termui.decorate(
ReportFormatter.TARGET_PREFIX, 'magenta', 'bold')
[docs] def format_activity(self, activity):
return termui.decorate(activity, *self.ACTIVITY_STYLES[activity])
[docs] def format_subject(self, subject, activity=None):
parent = super(ColoredReportFormatter, self)
subject = parent.format_subject(subject, activity)
return termui.decorate(subject, *self.SUBJECT_STYLES[activity])
[docs] def format_default(self, record):
record.msg = termui.decorate(
record.msg, *self.LOG_STYLES[record.levelname.lower()])
return super(ColoredReportFormatter, self).format_default(record)
[docs]class ReportLogger(LoggerAdapter):
"""Suitable wrapper for PyScaffold CLI interactive execution reports.
Args:
logger (logging.Logger): custom logger to be used. Optional: the
default logger will be used.
handlers (logging.Handler): custom logging handler to be used.
Optional: a :class:`logging.StreamHandler` is used by default.
formatter (logging.Formatter): custom formatter to be used.
Optional: by default a :class:`~.ReportFormatter` is created and
used.
extra (dict): extra attributes to be merged into the log record.
Options, empty by default.
Attributes:
wrapped (logging.Logger): underlying logger object.
handler (logging.Handler): stream handler configured for
providing user feedback in PyScaffold CLI.
formatter (logging.Formatter): formatter configured in the
default handler.
nesting (int): current nesting level of the report.
"""
def __init__(self, logger=None, handler=None, formatter=None, extra=None):
self.nesting = 0
self.wrapped = logger or getLogger(DEFAULT_LOGGER)
self.extra = extra or {}
self.handler = handler or StreamHandler()
self.formatter = formatter or ReportFormatter()
self.handler.setFormatter(self.formatter)
self.wrapped.addHandler(self.handler)
super(ReportLogger, self).__init__(self.wrapped, self.extra)
[docs] def process(self, msg, kwargs):
"""Method overridden to augment LogRecord with the `nesting` attribute.
"""
(msg, kwargs) = super(ReportLogger, self).process(msg, kwargs)
extra = kwargs.get('extra', {})
extra['nesting'] = self.nesting
kwargs['extra'] = extra
return msg, kwargs
[docs] def report(self, activity, subject,
context=None, target=None, nesting=None, level=INFO):
"""Log that an activity has occurred during scaffold.
Args:
activity (str): usually a verb or command, e.g. ``create``,
``invoke``, ``run``, ``chdir``...
subject (str): usually a path in the file system or an action
identifier.
context (str): path where the activity take place.
target (str): path affected by the activity
nesting (int): optional nesting level. By default it is calculated
from the activity name.
level (int): log level. Defaults to :obj:`logging.INFO`.
See :mod:`logging` for more information.
Notes:
This method creates a custom log record, with additional fields:
**activity**, **subject**, **context**, **target** and **nesting**,
but an empty **msg** field. The :class:`ReportFormatter`
creates the log message from the other fields.
Often **target** and **context** complement the logs when
**subject** does not hold all the necessary information. For
example::
logger.report('copy', 'my/file', target='my/awesome/path')
logger.report('run', 'command', context='current/working/dir')
"""
return self.wrapped.log(level, '', extra={
'activity': activity,
'subject': subject,
'context': context,
'target': target,
'nesting': nesting or self.nesting
})
[docs] @contextmanager
def indent(self, count=1):
"""Temporarily adjust padding while executing a context.
Example:
.. code-block:: python
from pyscaffold.log import logger
logger.report('invoke', 'custom_action')
with logger.indent():
logger.report('create', 'some/file/path')
# Expected logs:
# --------------------------------------
# invoke custom_action
# create some/file/path
# --------------------------------------
# Note how the spacing between activity and subject in the
# second entry is greater than the equivalent in the first one.
"""
prev = self.nesting
self.nesting += count
try:
yield
finally:
self.nesting = prev
[docs] def copy(self):
"""Produce a copy of the wrapped logger.
Sometimes, it is better to make a copy of th report logger to keep
indentation consistent.
"""
clone = self.__class__(self.wrapped, self.handler,
self.formatter, self.extra)
clone.nesting = self.nesting
return clone
logger = ReportLogger()
"""Default logger configured for PyScaffold."""