← Back
Editing: log.py
"""Logging """ import sys import os import logging from pip import backwardcompat from pip._vendor import colorama, pkg_resources def _color_wrap(*colors): def wrapped(inp): return "".join(list(colors) + [inp, colorama.Style.RESET_ALL]) return wrapped def should_color(consumer, environ, std=(sys.stdout, sys.stderr)): real_consumer = (consumer if not isinstance(consumer, colorama.AnsiToWin32) else consumer.wrapped) # If consumer isn't stdout or stderr we shouldn't colorize it if real_consumer not in std: return False # If consumer is a tty we should color it if hasattr(real_consumer, "isatty") and real_consumer.isatty(): return True # If we have an ASNI term we should color it if environ.get("TERM") == "ANSI": return True # If anything else we should not color it return False def should_warn(current_version, removal_version): # Our Significant digits on versions is 2, so remove everything but the # first two places. current_version = ".".join(current_version.split(".")[:2]) removal_version = ".".join(removal_version.split(".")[:2]) # Our warning threshold is one minor version before removal, so we # decrement the minor version by one major, minor = removal_version.split(".") minor = str(int(minor) - 1) warn_version = ".".join([major, minor]) # Test if our current_version should be a warn return (pkg_resources.parse_version(current_version) < pkg_resources.parse_version(warn_version)) class Logger(object): """ Logging object for use in command-line script. Allows ranges of levels, to avoid some redundancy of displayed information. """ VERBOSE_DEBUG = logging.DEBUG - 1 DEBUG = logging.DEBUG INFO = logging.INFO NOTIFY = (logging.INFO + logging.WARN) / 2 WARN = WARNING = logging.WARN ERROR = logging.ERROR FATAL = logging.FATAL LEVELS = [VERBOSE_DEBUG, DEBUG, INFO, NOTIFY, WARN, ERROR, FATAL] COLORS = { WARN: _color_wrap(colorama.Fore.YELLOW), ERROR: _color_wrap(colorama.Fore.RED), FATAL: _color_wrap(colorama.Fore.RED), } def __init__(self): self.consumers = [] self.indent = 0 self.explicit_levels = False self.in_progress = None self.in_progress_hanging = False def add_consumers(self, *consumers): if sys.platform.startswith("win"): for level, consumer in consumers: if hasattr(consumer, "write"): self.consumers.append( (level, colorama.AnsiToWin32(consumer)), ) else: self.consumers.append((level, consumer)) else: self.consumers.extend(consumers) def debug(self, msg, *args, **kw): self.log(self.DEBUG, msg, *args, **kw) def info(self, msg, *args, **kw): self.log(self.INFO, msg, *args, **kw) def notify(self, msg, *args, **kw): self.log(self.NOTIFY, msg, *args, **kw) def warn(self, msg, *args, **kw): self.log(self.WARN, msg, *args, **kw) def error(self, msg, *args, **kw): self.log(self.ERROR, msg, *args, **kw) def fatal(self, msg, *args, **kw): self.log(self.FATAL, msg, *args, **kw) def deprecated(self, removal_version, msg, *args, **kwargs): """ Logs deprecation message which is log level WARN if the ``removal_version`` is > 1 minor release away and log level ERROR otherwise. removal_version should be the version that the deprecated feature is expected to be removed in, so something that will not exist in version 1.7, but will in 1.6 would have a removal_version of 1.7. """ from pip import __version__ if should_warn(__version__, removal_version): self.warn(msg, *args, **kwargs) else: self.error(msg, *args, **kwargs) def log(self, level, msg, *args, **kw): if args: if kw: raise TypeError( "You may give positional or keyword arguments, not both") args = args or kw # render if args: rendered = msg % args else: rendered = msg rendered = ' ' * self.indent + rendered if self.explicit_levels: ## FIXME: should this be a name, not a level number? rendered = '%02i %s' % (level, rendered) for consumer_level, consumer in self.consumers: if self.level_matches(level, consumer_level): if (self.in_progress_hanging and consumer in (sys.stdout, sys.stderr)): self.in_progress_hanging = False sys.stdout.write('\n') sys.stdout.flush() if hasattr(consumer, 'write'): write_content = rendered + '\n' if should_color(consumer, os.environ): # We are printing to stdout or stderr and it supports # colors so render our text colored colorizer = self.COLORS.get(level, lambda x: x) write_content = colorizer(write_content) consumer.write(write_content) if hasattr(consumer, 'flush'): consumer.flush() else: consumer(rendered) def _show_progress(self): """Should we display download progress?""" return (self.stdout_level_matches(self.NOTIFY) and sys.stdout.isatty()) def start_progress(self, msg): assert not self.in_progress, ( "Tried to start_progress(%r) while in_progress %r" % (msg, self.in_progress)) if self._show_progress(): sys.stdout.write(' ' * self.indent + msg) sys.stdout.flush() self.in_progress_hanging = True else: self.in_progress_hanging = False self.in_progress = msg self.last_message = None def end_progress(self, msg='done.'): assert self.in_progress, ( "Tried to end_progress without start_progress") if self._show_progress(): if not self.in_progress_hanging: # Some message has been printed out since start_progress sys.stdout.write('...' + self.in_progress + msg + '\n') sys.stdout.flush() else: # These erase any messages shown with show_progress (besides .'s) logger.show_progress('') logger.show_progress('') sys.stdout.write(msg + '\n') sys.stdout.flush() self.in_progress = None self.in_progress_hanging = False def show_progress(self, message=None): """If we are in a progress scope, and no log messages have been shown, write out another '.'""" if self.in_progress_hanging: if message is None: sys.stdout.write('.') sys.stdout.flush() else: if self.last_message: padding = ' ' * max(0, len(self.last_message) - len(message)) else: padding = '' sys.stdout.write('\r%s%s%s%s' % (' ' * self.indent, self.in_progress, message, padding)) sys.stdout.flush() self.last_message = message def stdout_level_matches(self, level): """Returns true if a message at this level will go to stdout""" return self.level_matches(level, self._stdout_level()) def _stdout_level(self): """Returns the level that stdout runs at""" for level, consumer in self.consumers: if consumer is sys.stdout: return level return self.FATAL def level_matches(self, level, consumer_level): """ >>> l = Logger() >>> l.level_matches(3, 4) False >>> l.level_matches(3, 2) True >>> l.level_matches(slice(None, 3), 3) False >>> l.level_matches(slice(None, 3), 2) True >>> l.level_matches(slice(1, 3), 1) True >>> l.level_matches(slice(2, 3), 1) False """ if isinstance(level, slice): start, stop = level.start, level.stop if start is not None and start > consumer_level: return False if stop is not None or stop <= consumer_level: return False return True else: return level >= consumer_level @classmethod def level_for_integer(cls, level): levels = cls.LEVELS if level < 0: return levels[0] if level >= len(levels): return levels[-1] return levels[level] def move_stdout_to_stderr(self): to_remove = [] to_add = [] for consumer_level, consumer in self.consumers: if consumer == sys.stdout: to_remove.append((consumer_level, consumer)) to_add.append((consumer_level, sys.stderr)) for item in to_remove: self.consumers.remove(item) self.consumers.extend(to_add) logger = Logger()
Save File
Cancel