Source code for sphinx.util.logging

# -*- coding: utf-8 -*-

    Logging utility functions for Sphinx.

    :copyright: Copyright 2007-2018 by the Sphinx team, see AUTHORS.
    :license: BSD, see LICENSE for details.
from __future__ import absolute_import

import logging
import logging.handlers
from collections import defaultdict
from contextlib import contextmanager

from docutils import nodes
from docutils.utils import get_source_line
from six import PY2, StringIO

from sphinx.errors import SphinxWarning
from sphinx.util.console import colorize

if False:
    # For type annotation
    from typing import Any, Dict, Generator, IO, List, Tuple, Type, Union  # NOQA
    from docutils import nodes  # NOQA
    from sphinx.application import Sphinx  # NOQA

NAMESPACE = 'sphinx'

LEVEL_NAMES = defaultdict(lambda: logging.WARNING)  # type: Dict[str, int]
    'CRITICAL': logging.CRITICAL,
    'SEVERE': logging.CRITICAL,
    'ERROR': logging.ERROR,
    'WARNING': logging.WARNING,
    'INFO': logging.INFO,
    'DEBUG': logging.DEBUG,

VERBOSITY_MAP = defaultdict(lambda: 0)  # type: Dict[int, int]
    0: logging.INFO,
    1: VERBOSE,
    2: logging.DEBUG,

COLOR_MAP = defaultdict(lambda: 'blue')  # type: Dict[int, unicode]
    logging.ERROR: 'darkred',
    logging.WARNING: 'red',
    logging.DEBUG: 'darkgray',

[docs]def getLogger(name): # type: (str) -> SphinxLoggerAdapter """Get logger wrapped by :class:`sphinx.util.logging.SphinxLoggerAdapter`. Sphinx logger always uses ``sphinx.*`` namespace to be independent from settings of root logger. It ensures logging is consistent even if a third-party extension or imported application resets logger settings. Example usage:: >>> from sphinx.util import logging >>> logger = logging.getLogger(__name__) >>>'Hello, this is an extension!') Hello, this is an extension! """ # add sphinx prefix to name forcely logger = logging.getLogger(NAMESPACE + '.' + name) # Forcely enable logger logger.disabled = False # wrap logger by SphinxLoggerAdapter return SphinxLoggerAdapter(logger, {})
def convert_serializable(records): # type: (List[logging.LogRecord]) -> None """Convert LogRecord serializable.""" for r in records: # extract arguments to a message and clear them r.msg = r.getMessage() r.args = () location = getattr(r, 'location', None) if isinstance(location, nodes.Node): r.location = get_node_location(location) # type: ignore class SphinxLogRecord(logging.LogRecord): """Log record class supporting location""" prefix = '' location = None # type: Any def getMessage(self): # type: () -> str message = super(SphinxLogRecord, self).getMessage() location = getattr(self, 'location', None) if location: message = '%s: %s%s' % (location, self.prefix, message) elif self.prefix not in message: message = self.prefix + message return message class SphinxInfoLogRecord(SphinxLogRecord): """Info log record class supporting location""" prefix = '' # do not show any prefix for INFO messages class SphinxWarningLogRecord(SphinxLogRecord): """Warning log record class supporting location""" prefix = 'WARNING: '
[docs]class SphinxLoggerAdapter(logging.LoggerAdapter): """LoggerAdapter allowing ``type`` and ``subtype`` keywords."""
[docs] def log(self, level, msg, *args, **kwargs): # type: ignore # type: (Union[int, str], unicode, Any, Any) -> None if isinstance(level, int): super(SphinxLoggerAdapter, self).log(level, msg, *args, **kwargs) else: levelno = LEVEL_NAMES[level] super(SphinxLoggerAdapter, self).log(levelno, msg, *args, **kwargs)
[docs] def verbose(self, msg, *args, **kwargs): # type: (unicode, Any, Any) -> None self.log(VERBOSE, msg, *args, **kwargs)
def process(self, msg, kwargs): # type: ignore # type: (unicode, Dict) -> Tuple[unicode, Dict] extra = kwargs.setdefault('extra', {}) if 'type' in kwargs: extra['type'] = kwargs.pop('type') if 'subtype' in kwargs: extra['subtype'] = kwargs.pop('subtype') if 'location' in kwargs: extra['location'] = kwargs.pop('location') if 'nonl' in kwargs: extra['nonl'] = kwargs.pop('nonl') if 'color' in kwargs: extra['color'] = kwargs.pop('color') return msg, kwargs def handle(self, record): # type: (logging.LogRecord) -> None self.logger.handle(record) # type: ignore
class WarningStreamHandler(logging.StreamHandler): """StreamHandler for warnings.""" pass class NewLineStreamHandlerPY2(logging.StreamHandler): """StreamHandler which switches line terminator by record.nonl flag.""" def emit(self, record): # type: (logging.LogRecord) -> None try: self.acquire() stream = if getattr(record, 'nonl', False): # remove return code forcely when nonl=True = StringIO() super(NewLineStreamHandlerPY2, self).emit(record) stream.write([:-1]) stream.flush() else: super(NewLineStreamHandlerPY2, self).emit(record) finally: = stream self.release() class NewLineStreamHandlerPY3(logging.StreamHandler): """StreamHandler which switches line terminator by record.nonl flag.""" def emit(self, record): # type: (logging.LogRecord) -> None try: self.acquire() if getattr(record, 'nonl', False): # skip appending terminator when nonl=True self.terminator = '' super(NewLineStreamHandlerPY3, self).emit(record) finally: self.terminator = '\n' self.release() if PY2: NewLineStreamHandler = NewLineStreamHandlerPY2 else: NewLineStreamHandler = NewLineStreamHandlerPY3 class MemoryHandler(logging.handlers.BufferingHandler): """Handler buffering all logs.""" def __init__(self): # type: () -> None super(MemoryHandler, self).__init__(-1) def shouldFlush(self, record): # type: (logging.LogRecord) -> bool return False # never flush def flushTo(self, logger): # type: (logging.Logger) -> None self.acquire() try: for record in self.buffer: logger.handle(record) self.buffer = [] # type: List[logging.LogRecord] finally: self.release() def clear(self): # type: () -> List[logging.LogRecord] buffer, self.buffer = self.buffer, [] return buffer
[docs]@contextmanager def pending_warnings(): # type: () -> Generator """Contextmanager to pend logging warnings temporary. Similar to :func:`pending_logging`. """ logger = logging.getLogger(NAMESPACE) memhandler = MemoryHandler() memhandler.setLevel(logging.WARNING) try: handlers = [] for handler in logger.handlers[:]: if isinstance(handler, WarningStreamHandler): logger.removeHandler(handler) handlers.append(handler) logger.addHandler(memhandler) yield memhandler finally: logger.removeHandler(memhandler) for handler in handlers: logger.addHandler(handler) memhandler.flushTo(logger)
[docs]@contextmanager def pending_logging(): # type: () -> Generator """Contextmanager to pend logging all logs temporary. For example:: >>> with pending_logging(): >>> logger.warning('Warning message!') # not flushed yet >>> some_long_process() >>> Warning message! # the warning is flushed here """ logger = logging.getLogger(NAMESPACE) memhandler = MemoryHandler() try: handlers = [] for handler in logger.handlers[:]: logger.removeHandler(handler) handlers.append(handler) logger.addHandler(memhandler) yield memhandler finally: logger.removeHandler(memhandler) for handler in handlers: logger.addHandler(handler) memhandler.flushTo(logger)
@contextmanager def skip_warningiserror(skip=True): # type: (bool) -> Generator """contextmanager to skip WarningIsErrorFilter for a while.""" logger = logging.getLogger(NAMESPACE) if skip is False: yield else: try: disabler = DisableWarningIsErrorFilter() for handler in logger.handlers: # use internal method; filters.insert() directly to install disabler # before WarningIsErrorFilter handler.filters.insert(0, disabler) yield finally: for handler in logger.handlers: handler.removeFilter(disabler) class LogCollector(object): def __init__(self): # type: () -> None self.logs = [] # type: List[logging.LogRecord] @contextmanager def collect(self): # type: () -> Generator with pending_logging() as memhandler: yield self.logs = memhandler.clear() class InfoFilter(logging.Filter): """Filter error and warning messages.""" def filter(self, record): # type: (logging.LogRecord) -> bool if record.levelno < logging.WARNING: return True else: return False def is_suppressed_warning(type, subtype, suppress_warnings): # type: (unicode, unicode, List[unicode]) -> bool """Check the warning is suppressed or not.""" if type is None: return False for warning_type in suppress_warnings: if '.' in warning_type: target, subtarget = warning_type.split('.', 1) else: target, subtarget = warning_type, None if target == type: if (subtype is None or subtarget is None or subtarget == subtype or subtarget == '*'): return True return False class WarningSuppressor(logging.Filter): """Filter logs by `suppress_warnings`.""" def __init__(self, app): # type: (Sphinx) -> None = app super(WarningSuppressor, self).__init__() def filter(self, record): # type: (logging.LogRecord) -> bool type = getattr(record, 'type', None) subtype = getattr(record, 'subtype', None) try: suppress_warnings = except AttributeError: # config is not initialized yet (ex. in suppress_warnings = [] if is_suppressed_warning(type, subtype, suppress_warnings): return False else: += 1 return True class WarningIsErrorFilter(logging.Filter): """Raise exception if warning emitted.""" def __init__(self, app): # type: (Sphinx) -> None = app super(WarningIsErrorFilter, self).__init__() def filter(self, record): # type: (logging.LogRecord) -> bool if getattr(record, 'skip_warningsiserror', False): # disabled by DisableWarningIsErrorFilter return True elif location = getattr(record, 'location', '') try: message = record.msg % record.args except (TypeError, ValueError): message = record.msg # use record.msg itself if location: raise SphinxWarning(location + ":" + message) else: raise SphinxWarning(message) else: return True class DisableWarningIsErrorFilter(logging.Filter): """Disable WarningIsErrorFilter if this filter installed.""" def filter(self, record): # type: (logging.LogRecord) -> bool record.skip_warningsiserror = True # type: ignore return True class SphinxLogRecordTranslator(logging.Filter): """Converts a log record to one Sphinx expects * Make a instance of SphinxLogRecord * docname to path if location given """ LogRecordClass = None # type: Type[logging.LogRecord] def __init__(self, app): # type: (Sphinx) -> None = app super(SphinxLogRecordTranslator, self).__init__() def filter(self, record): # type: ignore # type: (SphinxWarningLogRecord) -> bool if isinstance(record, logging.LogRecord): record.__class__ = self.LogRecordClass # force subclassing to handle location location = getattr(record, 'location', None) if isinstance(location, tuple): docname, lineno = location if docname and lineno: record.location = '%s:%s' % (, lineno) elif docname: record.location = '%s' % else: record.location = None elif isinstance(location, nodes.Node): record.location = get_node_location(location) elif location and ':' not in location: record.location = '%s' % return True class InfoLogRecordTranslator(SphinxLogRecordTranslator): """LogRecordTranslator for INFO level log records.""" LogRecordClass = SphinxInfoLogRecord class WarningLogRecordTranslator(SphinxLogRecordTranslator): """LogRecordTranslator for WARNING level log records.""" LogRecordClass = SphinxWarningLogRecord def get_node_location(node): # type: (nodes.Node) -> str (source, line) = get_source_line(node) if source and line: return "%s:%s" % (source, line) elif source: return "%s:" % source elif line: return "<unknown>:%s" % line else: return None class ColorizeFormatter(logging.Formatter): def format(self, record): # type: (logging.LogRecord) -> str message = super(ColorizeFormatter, self).format(record) color = getattr(record, 'color', None) if color is None: color = COLOR_MAP.get(record.levelno) if color: return colorize(color, message) # type: ignore else: return message class SafeEncodingWriter(object): """Stream writer which ignores UnicodeEncodeError silently""" def __init__(self, stream): # type: (IO) -> None = stream self.encoding = getattr(stream, 'encoding', 'ascii') or 'ascii' def write(self, data): # type: (unicode) -> None try: except UnicodeEncodeError: # stream accept only str, not bytes. So, we encode and replace # non-encodable characters, then decode them., 'replace').decode(self.encoding)) def flush(self): # type: () -> None if hasattr(, 'flush'): class LastMessagesWriter(object): """Stream writer which memories last 10 messages to save trackback""" def __init__(self, app, stream): # type: (Sphinx, IO) -> None = app def write(self, data): # type: (unicode) -> None def setup(app, status, warning): # type: (Sphinx, IO, IO) -> None """Setup root logger for Sphinx""" logger = logging.getLogger(NAMESPACE) logger.setLevel(logging.DEBUG) logger.propagate = False # clear all handlers for handler in logger.handlers[:]: logger.removeHandler(handler) info_handler = NewLineStreamHandler(SafeEncodingWriter(status)) # type: ignore info_handler.addFilter(InfoFilter()) info_handler.addFilter(InfoLogRecordTranslator(app)) info_handler.setLevel(VERBOSITY_MAP[app.verbosity]) info_handler.setFormatter(ColorizeFormatter()) warning_handler = WarningStreamHandler(SafeEncodingWriter(warning)) # type: ignore warning_handler.addFilter(WarningSuppressor(app)) warning_handler.addFilter(WarningLogRecordTranslator(app)) warning_handler.addFilter(WarningIsErrorFilter(app)) warning_handler.setLevel(logging.WARNING) warning_handler.setFormatter(ColorizeFormatter()) messagelog_handler = logging.StreamHandler(LastMessagesWriter(app, status)) # type: ignore messagelog_handler.addFilter(InfoFilter()) messagelog_handler.setLevel(VERBOSITY_MAP[app.verbosity]) messagelog_handler.setFormatter(ColorizeFormatter()) logger.addHandler(info_handler) logger.addHandler(warning_handler) logger.addHandler(messagelog_handler)