123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182 |
- from __future__ import absolute_import
- import logging
- import os
- import textwrap
- import pip._internal.utils.filesystem as filesystem
- from pip._internal.cli.base_command import Command
- from pip._internal.cli.status_codes import ERROR, SUCCESS
- from pip._internal.exceptions import CommandError, PipError
- from pip._internal.utils.typing import MYPY_CHECK_RUNNING
- if MYPY_CHECK_RUNNING:
- from optparse import Values
- from typing import Any, List
- logger = 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)
|