diff options
Diffstat (limited to 'sphinx/util/logging.py')
| -rw-r--r-- | sphinx/util/logging.py | 116 |
1 files changed, 40 insertions, 76 deletions
diff --git a/sphinx/util/logging.py b/sphinx/util/logging.py index afa4ebd23..d6667dacd 100644 --- a/sphinx/util/logging.py +++ b/sphinx/util/logging.py @@ -12,8 +12,10 @@ import logging import logging.handlers from collections import defaultdict from contextlib import contextmanager +from typing import Any, Dict, Generator, IO, List, Tuple, Type, Union from docutils import nodes +from docutils.nodes import Node from docutils.utils import get_source_line from sphinx.errors import SphinxWarning @@ -21,8 +23,7 @@ from sphinx.util.console import colorize if False: # For type annotation - from typing import Any, Dict, Generator, IO, List, Tuple, Type, Union # NOQA - from sphinx.application import Sphinx # NOQA + from sphinx.application import Sphinx NAMESPACE = 'sphinx' @@ -54,8 +55,7 @@ COLOR_MAP = defaultdict(lambda: 'blue', }) -def getLogger(name): - # type: (str) -> SphinxLoggerAdapter +def getLogger(name: str) -> "SphinxLoggerAdapter": """Get logger wrapped by :class:`sphinx.util.logging.SphinxLoggerAdapter`. Sphinx logger always uses ``sphinx.*`` namespace to be independent from @@ -77,8 +77,7 @@ def getLogger(name): return SphinxLoggerAdapter(logger, {}) -def convert_serializable(records): - # type: (List[logging.LogRecord]) -> None +def convert_serializable(records: List[logging.LogRecord]) -> None: """Convert LogRecord serializable.""" for r in records: # extract arguments to a message and clear them @@ -95,8 +94,7 @@ class SphinxLogRecord(logging.LogRecord): prefix = '' location = None # type: Any - def getMessage(self): - # type: () -> str + def getMessage(self) -> str: message = super().getMessage() location = getattr(self, 'location', None) if location: @@ -120,20 +118,17 @@ class SphinxWarningLogRecord(SphinxLogRecord): class SphinxLoggerAdapter(logging.LoggerAdapter): """LoggerAdapter allowing ``type`` and ``subtype`` keywords.""" - def log(self, level, msg, *args, **kwargs): - # type: (Union[int, str], str, Any, Any) -> None + def log(self, level: Union[int, str], msg: str, *args, **kwargs) -> None: if isinstance(level, int): super().log(level, msg, *args, **kwargs) else: levelno = LEVEL_NAMES[level] super().log(levelno, msg, *args, **kwargs) - def verbose(self, msg, *args, **kwargs): - # type: (str, Any, Any) -> None + def verbose(self, msg: str, *args, **kwargs) -> None: self.log(VERBOSE, msg, *args, **kwargs) - def process(self, msg, kwargs): # type: ignore - # type: (str, Dict) -> Tuple[str, Dict] + def process(self, msg: str, kwargs: Dict) -> Tuple[str, Dict]: # type: ignore extra = kwargs.setdefault('extra', {}) if 'type' in kwargs: extra['type'] = kwargs.pop('type') @@ -148,8 +143,7 @@ class SphinxLoggerAdapter(logging.LoggerAdapter): return msg, kwargs - def handle(self, record): - # type: (logging.LogRecord) -> None + def handle(self, record: logging.LogRecord) -> None: self.logger.handle(record) @@ -161,8 +155,7 @@ class WarningStreamHandler(logging.StreamHandler): class NewLineStreamHandler(logging.StreamHandler): """StreamHandler which switches line terminator by record.nonl flag.""" - def emit(self, record): - # type: (logging.LogRecord) -> None + def emit(self, record: logging.LogRecord) -> None: try: self.acquire() if getattr(record, 'nonl', False): @@ -177,16 +170,13 @@ class NewLineStreamHandler(logging.StreamHandler): class MemoryHandler(logging.handlers.BufferingHandler): """Handler buffering all logs.""" - def __init__(self): - # type: () -> None + def __init__(self) -> None: super().__init__(-1) - def shouldFlush(self, record): - # type: (logging.LogRecord) -> bool + def shouldFlush(self, record: logging.LogRecord) -> bool: return False # never flush - def flushTo(self, logger): - # type: (logging.Logger) -> None + def flushTo(self, logger: logging.Logger) -> None: self.acquire() try: for record in self.buffer: @@ -195,15 +185,13 @@ class MemoryHandler(logging.handlers.BufferingHandler): finally: self.release() - def clear(self): - # type: () -> List[logging.LogRecord] + def clear(self) -> List[logging.LogRecord]: buffer, self.buffer = self.buffer, [] return buffer @contextmanager -def pending_warnings(): - # type: () -> Generator +def pending_warnings() -> Generator[logging.Handler, None, None]: """Contextmanager to pend logging warnings temporary. Similar to :func:`pending_logging`. @@ -231,8 +219,7 @@ def pending_warnings(): @contextmanager -def pending_logging(): - # type: () -> Generator +def pending_logging() -> Generator[MemoryHandler, None, None]: """Contextmanager to pend logging all logs temporary. For example:: @@ -264,8 +251,7 @@ def pending_logging(): @contextmanager -def skip_warningiserror(skip=True): - # type: (bool) -> Generator +def skip_warningiserror(skip: bool = True) -> Generator[None, None, None]: """contextmanager to skip WarningIsErrorFilter for a while.""" logger = logging.getLogger(NAMESPACE) @@ -285,8 +271,7 @@ def skip_warningiserror(skip=True): @contextmanager -def prefixed_warnings(prefix): - # type: (str) -> Generator +def prefixed_warnings(prefix: str) -> Generator[None, None, None]: """Prepend prefix to all records for a while. For example:: @@ -332,13 +317,11 @@ def prefixed_warnings(prefix): class LogCollector: - def __init__(self): - # type: () -> None + def __init__(self) -> None: self.logs = [] # type: List[logging.LogRecord] @contextmanager - def collect(self): - # type: () -> Generator + def collect(self) -> Generator[None, None, None]: with pending_logging() as memhandler: yield @@ -348,16 +331,14 @@ class LogCollector: class InfoFilter(logging.Filter): """Filter error and warning messages.""" - def filter(self, record): - # type: (logging.LogRecord) -> bool + def filter(self, record: logging.LogRecord) -> bool: if record.levelno < logging.WARNING: return True else: return False -def is_suppressed_warning(type, subtype, suppress_warnings): - # type: (str, str, List[str]) -> bool +def is_suppressed_warning(type: str, subtype: str, suppress_warnings: List[str]) -> bool: """Check the warning is suppressed or not.""" if type is None: return False @@ -379,13 +360,11 @@ def is_suppressed_warning(type, subtype, suppress_warnings): class WarningSuppressor(logging.Filter): """Filter logs by `suppress_warnings`.""" - def __init__(self, app): - # type: (Sphinx) -> None + def __init__(self, app: "Sphinx") -> None: self.app = app super().__init__() - def filter(self, record): - # type: (logging.LogRecord) -> bool + def filter(self, record: logging.LogRecord) -> bool: type = getattr(record, 'type', None) subtype = getattr(record, 'subtype', None) @@ -405,13 +384,11 @@ class WarningSuppressor(logging.Filter): class WarningIsErrorFilter(logging.Filter): """Raise exception if warning emitted.""" - def __init__(self, app): - # type: (Sphinx) -> None + def __init__(self, app: "Sphinx") -> None: self.app = app super().__init__() - def filter(self, record): - # type: (logging.LogRecord) -> bool + def filter(self, record: logging.LogRecord) -> bool: if getattr(record, 'skip_warningsiserror', False): # disabled by DisableWarningIsErrorFilter return True @@ -433,8 +410,7 @@ class WarningIsErrorFilter(logging.Filter): class DisableWarningIsErrorFilter(logging.Filter): """Disable WarningIsErrorFilter if this filter installed.""" - def filter(self, record): - # type: (logging.LogRecord) -> bool + def filter(self, record: logging.LogRecord) -> bool: record.skip_warningsiserror = True # type: ignore return True @@ -442,13 +418,11 @@ class DisableWarningIsErrorFilter(logging.Filter): class MessagePrefixFilter(logging.Filter): """Prepend prefix to all records.""" - def __init__(self, prefix): - # type: (str) -> None + def __init__(self, prefix: str) -> None: self.prefix = prefix super().__init__() - def filter(self, record): - # type: (logging.LogRecord) -> bool + def filter(self, record: logging.LogRecord) -> bool: if self.prefix: record.msg = self.prefix + ' ' + record.msg return True @@ -462,13 +436,11 @@ class SphinxLogRecordTranslator(logging.Filter): """ LogRecordClass = None # type: Type[logging.LogRecord] - def __init__(self, app): - # type: (Sphinx) -> None + def __init__(self, app: "Sphinx") -> None: self.app = app super().__init__() - def filter(self, record): # type: ignore - # type: (SphinxWarningLogRecord) -> bool + def filter(self, record: SphinxWarningLogRecord) -> bool: # type: ignore if isinstance(record, logging.LogRecord): # force subclassing to handle location record.__class__ = self.LogRecordClass # type: ignore @@ -500,8 +472,7 @@ class WarningLogRecordTranslator(SphinxLogRecordTranslator): LogRecordClass = SphinxWarningLogRecord -def get_node_location(node): - # type: (nodes.Node) -> str +def get_node_location(node: Node) -> str: (source, line) = get_source_line(node) if source and line: return "%s:%s" % (source, line) @@ -514,8 +485,7 @@ def get_node_location(node): class ColorizeFormatter(logging.Formatter): - def format(self, record): - # type: (logging.LogRecord) -> str + def format(self, record: logging.LogRecord) -> str: message = super().format(record) color = getattr(record, 'color', None) if color is None: @@ -529,13 +499,11 @@ class ColorizeFormatter(logging.Formatter): class SafeEncodingWriter: """Stream writer which ignores UnicodeEncodeError silently""" - def __init__(self, stream): - # type: (IO) -> None + def __init__(self, stream: IO) -> None: self.stream = stream self.encoding = getattr(stream, 'encoding', 'ascii') or 'ascii' - def write(self, data): - # type: (str) -> None + def write(self, data: str) -> None: try: self.stream.write(data) except UnicodeEncodeError: @@ -543,25 +511,21 @@ class SafeEncodingWriter: # non-encodable characters, then decode them. self.stream.write(data.encode(self.encoding, 'replace').decode(self.encoding)) - def flush(self): - # type: () -> None + def flush(self) -> None: if hasattr(self.stream, 'flush'): self.stream.flush() class LastMessagesWriter: """Stream writer which memories last 10 messages to save trackback""" - def __init__(self, app, stream): - # type: (Sphinx, IO) -> None + def __init__(self, app: "Sphinx", stream: IO) -> None: self.app = app - def write(self, data): - # type: (str) -> None + def write(self, data: str) -> None: self.app.messagelog.append(data) -def setup(app, status, warning): - # type: (Sphinx, IO, IO) -> None +def setup(app: "Sphinx", status: IO, warning: IO) -> None: """Setup root logger for Sphinx""" logger = logging.getLogger(NAMESPACE) logger.setLevel(logging.DEBUG) |