| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182 | from __future__ import absolute_importimport loggingimport osimport textwrapimport pip._internal.utils.filesystem as filesystemfrom pip._internal.cli.base_command import Commandfrom pip._internal.cli.status_codes import ERROR, SUCCESSfrom pip._internal.exceptions import CommandError, PipErrorfrom pip._internal.utils.typing import MYPY_CHECK_RUNNINGif MYPY_CHECK_RUNNING:    from optparse import Values    from typing import Any, Listlogger = logging.getLogger(__name__)class CacheCommand(Command):    """    Inspect and manage pip's wheel cache.    Subcommands:    - dir: Show the cache directory.    - info: Show information about the cache.    - list: List filenames of packages stored in the cache.    - remove: Remove one or more package from the cache.    - purge: Remove all items from the cache.    ``<pattern>`` can be a glob expression or a package name.    """    ignore_require_venv = True    usage = """        %prog dir        %prog info        %prog list [<pattern>]        %prog remove <pattern>        %prog purge    """    def run(self, options, args):        # type: (Values, List[Any]) -> int        handlers = {            "dir": self.get_cache_dir,            "info": self.get_cache_info,            "list": self.list_cache_items,            "remove": self.remove_cache_items,            "purge": self.purge_cache,        }        if not options.cache_dir:            logger.error("pip cache commands can not "                         "function since cache is disabled.")            return ERROR        # Determine action        if not args or args[0] not in handlers:            logger.error(                "Need an action (%s) to perform.",                ", ".join(sorted(handlers)),            )            return ERROR        action = args[0]        # Error handling happens here, not in the action-handlers.        try:            handlers[action](options, args[1:])        except PipError as e:            logger.error(e.args[0])            return ERROR        return SUCCESS    def get_cache_dir(self, options, args):        # type: (Values, List[Any]) -> None        if args:            raise CommandError('Too many arguments')        logger.info(options.cache_dir)    def get_cache_info(self, options, args):        # type: (Values, List[Any]) -> None        if args:            raise CommandError('Too many arguments')        num_packages = len(self._find_wheels(options, '*'))        cache_location = self._wheels_cache_dir(options)        cache_size = filesystem.format_directory_size(cache_location)        message = textwrap.dedent("""            Location: {location}            Size: {size}            Number of wheels: {package_count}        """).format(            location=cache_location,            package_count=num_packages,            size=cache_size,        ).strip()        logger.info(message)    def list_cache_items(self, options, args):        # type: (Values, List[Any]) -> None        if len(args) > 1:            raise CommandError('Too many arguments')        if args:            pattern = args[0]        else:            pattern = '*'        files = self._find_wheels(options, pattern)        if not files:            logger.info('Nothing cached.')            return        results = []        for filename in files:            wheel = os.path.basename(filename)            size = filesystem.format_file_size(filename)            results.append(' - {} ({})'.format(wheel, size))        logger.info('Cache contents:\n')        logger.info('\n'.join(sorted(results)))    def remove_cache_items(self, options, args):        # type: (Values, List[Any]) -> None        if len(args) > 1:            raise CommandError('Too many arguments')        if not args:            raise CommandError('Please provide a pattern')        files = self._find_wheels(options, args[0])        if not files:            raise CommandError('No matching packages')        for filename in files:            os.unlink(filename)            logger.debug('Removed %s', filename)        logger.info('Files removed: %s', len(files))    def purge_cache(self, options, args):        # type: (Values, List[Any]) -> None        if args:            raise CommandError('Too many arguments')        return self.remove_cache_items(options, ['*'])    def _wheels_cache_dir(self, options):        # type: (Values) -> str        return os.path.join(options.cache_dir, 'wheels')    def _find_wheels(self, options, pattern):        # type: (Values, str) -> List[str]        wheel_dir = self._wheels_cache_dir(options)        # The wheel filename format, as specified in PEP 427, is:        #     {distribution}-{version}(-{build})?-{python}-{abi}-{platform}.whl        #        # Additionally, non-alphanumeric values in the distribution are        # normalized to underscores (_), meaning hyphens can never occur        # before `-{version}`.        #        # Given that information:        # - If the pattern we're given contains a hyphen (-), the user is        #   providing at least the version. Thus, we can just append `*.whl`        #   to match the rest of it.        # - If the pattern we're given doesn't contain a hyphen (-), the        #   user is only providing the name. Thus, we append `-*.whl` to        #   match the hyphen before the version, followed by anything else.        #        # PEP 427: https://www.python.org/dev/peps/pep-0427/        pattern = pattern + ("*.whl" if "-" in pattern else "-*.whl")        return filesystem.find_files(wheel_dir, pattern)
 |