diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/pip/_internal/utils/logging.py')
| -rw-r--r-- | venv/lib/python3.11/site-packages/pip/_internal/utils/logging.py | 348 |
1 files changed, 0 insertions, 348 deletions
diff --git a/venv/lib/python3.11/site-packages/pip/_internal/utils/logging.py b/venv/lib/python3.11/site-packages/pip/_internal/utils/logging.py deleted file mode 100644 index 95982df..0000000 --- a/venv/lib/python3.11/site-packages/pip/_internal/utils/logging.py +++ /dev/null @@ -1,348 +0,0 @@ -import contextlib -import errno -import logging -import logging.handlers -import os -import sys -import threading -from dataclasses import dataclass -from io import TextIOWrapper -from logging import Filter -from typing import Any, ClassVar, Generator, List, Optional, TextIO, Type - -from pip._vendor.rich.console import ( - Console, - ConsoleOptions, - ConsoleRenderable, - RenderableType, - RenderResult, - RichCast, -) -from pip._vendor.rich.highlighter import NullHighlighter -from pip._vendor.rich.logging import RichHandler -from pip._vendor.rich.segment import Segment -from pip._vendor.rich.style import Style - -from pip._internal.utils._log import VERBOSE, getLogger -from pip._internal.utils.compat import WINDOWS -from pip._internal.utils.deprecation import DEPRECATION_MSG_PREFIX -from pip._internal.utils.misc import ensure_dir - -_log_state = threading.local() -subprocess_logger = getLogger("pip.subprocessor") - - -class BrokenStdoutLoggingError(Exception): - """ - Raised if BrokenPipeError occurs for the stdout stream while logging. - """ - - -def _is_broken_pipe_error(exc_class: Type[BaseException], exc: BaseException) -> bool: - if exc_class is BrokenPipeError: - return True - - # On Windows, a broken pipe can show up as EINVAL rather than EPIPE: - # https://bugs.python.org/issue19612 - # https://bugs.python.org/issue30418 - if not WINDOWS: - return False - - return isinstance(exc, OSError) and exc.errno in (errno.EINVAL, errno.EPIPE) - - -@contextlib.contextmanager -def indent_log(num: int = 2) -> Generator[None, None, None]: - """ - A context manager which will cause the log output to be indented for any - log messages emitted inside it. - """ - # For thread-safety - _log_state.indentation = get_indentation() - _log_state.indentation += num - try: - yield - finally: - _log_state.indentation -= num - - -def get_indentation() -> int: - return getattr(_log_state, "indentation", 0) - - -class IndentingFormatter(logging.Formatter): - default_time_format = "%Y-%m-%dT%H:%M:%S" - - def __init__( - self, - *args: Any, - add_timestamp: bool = False, - **kwargs: Any, - ) -> None: - """ - A logging.Formatter that obeys the indent_log() context manager. - - :param add_timestamp: A bool indicating output lines should be prefixed - with their record's timestamp. - """ - self.add_timestamp = add_timestamp - super().__init__(*args, **kwargs) - - def get_message_start(self, formatted: str, levelno: int) -> str: - """ - Return the start of the formatted log message (not counting the - prefix to add to each line). - """ - if levelno < logging.WARNING: - return "" - if formatted.startswith(DEPRECATION_MSG_PREFIX): - # Then the message already has a prefix. We don't want it to - # look like "WARNING: DEPRECATION: ...." - return "" - if levelno < logging.ERROR: - return "WARNING: " - - return "ERROR: " - - def format(self, record: logging.LogRecord) -> str: - """ - Calls the standard formatter, but will indent all of the log message - lines by our current indentation level. - """ - formatted = super().format(record) - message_start = self.get_message_start(formatted, record.levelno) - formatted = message_start + formatted - - prefix = "" - if self.add_timestamp: - prefix = f"{self.formatTime(record)} " - prefix += " " * get_indentation() - formatted = "".join([prefix + line for line in formatted.splitlines(True)]) - return formatted - - -@dataclass -class IndentedRenderable: - renderable: RenderableType - indent: int - - def __rich_console__( - self, console: Console, options: ConsoleOptions - ) -> RenderResult: - segments = console.render(self.renderable, options) - lines = Segment.split_lines(segments) - for line in lines: - yield Segment(" " * self.indent) - yield from line - yield Segment("\n") - - -class RichPipStreamHandler(RichHandler): - KEYWORDS: ClassVar[Optional[List[str]]] = [] - - def __init__(self, stream: Optional[TextIO], no_color: bool) -> None: - super().__init__( - console=Console(file=stream, no_color=no_color, soft_wrap=True), - show_time=False, - show_level=False, - show_path=False, - highlighter=NullHighlighter(), - ) - - # Our custom override on Rich's logger, to make things work as we need them to. - def emit(self, record: logging.LogRecord) -> None: - style: Optional[Style] = None - - # If we are given a diagnostic error to present, present it with indentation. - assert isinstance(record.args, tuple) - if getattr(record, "rich", False): - (rich_renderable,) = record.args - assert isinstance( - rich_renderable, (ConsoleRenderable, RichCast, str) - ), f"{rich_renderable} is not rich-console-renderable" - - renderable: RenderableType = IndentedRenderable( - rich_renderable, indent=get_indentation() - ) - else: - message = self.format(record) - renderable = self.render_message(record, message) - if record.levelno is not None: - if record.levelno >= logging.ERROR: - style = Style(color="red") - elif record.levelno >= logging.WARNING: - style = Style(color="yellow") - - try: - self.console.print(renderable, overflow="ignore", crop=False, style=style) - except Exception: - self.handleError(record) - - def handleError(self, record: logging.LogRecord) -> None: - """Called when logging is unable to log some output.""" - - exc_class, exc = sys.exc_info()[:2] - # If a broken pipe occurred while calling write() or flush() on the - # stdout stream in logging's Handler.emit(), then raise our special - # exception so we can handle it in main() instead of logging the - # broken pipe error and continuing. - if ( - exc_class - and exc - and self.console.file is sys.stdout - and _is_broken_pipe_error(exc_class, exc) - ): - raise BrokenStdoutLoggingError() - - return super().handleError(record) - - -class BetterRotatingFileHandler(logging.handlers.RotatingFileHandler): - def _open(self) -> TextIOWrapper: - ensure_dir(os.path.dirname(self.baseFilename)) - return super()._open() - - -class MaxLevelFilter(Filter): - def __init__(self, level: int) -> None: - self.level = level - - def filter(self, record: logging.LogRecord) -> bool: - return record.levelno < self.level - - -class ExcludeLoggerFilter(Filter): - - """ - A logging Filter that excludes records from a logger (or its children). - """ - - def filter(self, record: logging.LogRecord) -> bool: - # The base Filter class allows only records from a logger (or its - # children). - return not super().filter(record) - - -def setup_logging(verbosity: int, no_color: bool, user_log_file: Optional[str]) -> int: - """Configures and sets up all of the logging - - Returns the requested logging level, as its integer value. - """ - - # Determine the level to be logging at. - if verbosity >= 2: - level_number = logging.DEBUG - elif verbosity == 1: - level_number = VERBOSE - elif verbosity == -1: - level_number = logging.WARNING - elif verbosity == -2: - level_number = logging.ERROR - elif verbosity <= -3: - level_number = logging.CRITICAL - else: - level_number = logging.INFO - - level = logging.getLevelName(level_number) - - # The "root" logger should match the "console" level *unless* we also need - # to log to a user log file. - include_user_log = user_log_file is not None - if include_user_log: - additional_log_file = user_log_file - root_level = "DEBUG" - else: - additional_log_file = "/dev/null" - root_level = level - - # Disable any logging besides WARNING unless we have DEBUG level logging - # enabled for vendored libraries. - vendored_log_level = "WARNING" if level in ["INFO", "ERROR"] else "DEBUG" - - # Shorthands for clarity - log_streams = { - "stdout": "ext://sys.stdout", - "stderr": "ext://sys.stderr", - } - handler_classes = { - "stream": "pip._internal.utils.logging.RichPipStreamHandler", - "file": "pip._internal.utils.logging.BetterRotatingFileHandler", - } - handlers = ["console", "console_errors", "console_subprocess"] + ( - ["user_log"] if include_user_log else [] - ) - - logging.config.dictConfig( - { - "version": 1, - "disable_existing_loggers": False, - "filters": { - "exclude_warnings": { - "()": "pip._internal.utils.logging.MaxLevelFilter", - "level": logging.WARNING, - }, - "restrict_to_subprocess": { - "()": "logging.Filter", - "name": subprocess_logger.name, - }, - "exclude_subprocess": { - "()": "pip._internal.utils.logging.ExcludeLoggerFilter", - "name": subprocess_logger.name, - }, - }, - "formatters": { - "indent": { - "()": IndentingFormatter, - "format": "%(message)s", - }, - "indent_with_timestamp": { - "()": IndentingFormatter, - "format": "%(message)s", - "add_timestamp": True, - }, - }, - "handlers": { - "console": { - "level": level, - "class": handler_classes["stream"], - "no_color": no_color, - "stream": log_streams["stdout"], - "filters": ["exclude_subprocess", "exclude_warnings"], - "formatter": "indent", - }, - "console_errors": { - "level": "WARNING", - "class": handler_classes["stream"], - "no_color": no_color, - "stream": log_streams["stderr"], - "filters": ["exclude_subprocess"], - "formatter": "indent", - }, - # A handler responsible for logging to the console messages - # from the "subprocessor" logger. - "console_subprocess": { - "level": level, - "class": handler_classes["stream"], - "stream": log_streams["stderr"], - "no_color": no_color, - "filters": ["restrict_to_subprocess"], - "formatter": "indent", - }, - "user_log": { - "level": "DEBUG", - "class": handler_classes["file"], - "filename": additional_log_file, - "encoding": "utf-8", - "delay": True, - "formatter": "indent_with_timestamp", - }, - }, - "root": { - "level": root_level, - "handlers": handlers, - }, - "loggers": {"pip._vendor": {"level": vendored_log_level}}, - } - ) - - return level_number |