Spaces:
Running
Running
| import os | |
| import textwrap | |
| from optparse import Values | |
| from typing import Any, List | |
| 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.logging import getLogger | |
| logger = 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>] [--format=[human, abspath]] | |
| %prog remove <pattern> | |
| %prog purge | |
| """ | |
| def add_options(self) -> None: | |
| self.cmd_opts.add_option( | |
| "--format", | |
| action="store", | |
| dest="list_format", | |
| default="human", | |
| choices=("human", "abspath"), | |
| help="Select the output format among: human (default) or abspath", | |
| ) | |
| self.parser.insert_option_group(0, self.cmd_opts) | |
| def run(self, options: Values, args: List[str]) -> 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: Values, args: List[Any]) -> None: | |
| if args: | |
| raise CommandError("Too many arguments") | |
| logger.info(options.cache_dir) | |
| def get_cache_info(self, options: Values, args: List[Any]) -> None: | |
| if args: | |
| raise CommandError("Too many arguments") | |
| num_http_files = len(self._find_http_files(options)) | |
| num_packages = len(self._find_wheels(options, "*")) | |
| http_cache_location = self._cache_dir(options, "http") | |
| wheels_cache_location = self._cache_dir(options, "wheels") | |
| http_cache_size = filesystem.format_directory_size(http_cache_location) | |
| wheels_cache_size = filesystem.format_directory_size(wheels_cache_location) | |
| message = ( | |
| textwrap.dedent( | |
| """ | |
| Package index page cache location: {http_cache_location} | |
| Package index page cache size: {http_cache_size} | |
| Number of HTTP files: {num_http_files} | |
| Locally built wheels location: {wheels_cache_location} | |
| Locally built wheels size: {wheels_cache_size} | |
| Number of locally built wheels: {package_count} | |
| """ | |
| ) | |
| .format( | |
| http_cache_location=http_cache_location, | |
| http_cache_size=http_cache_size, | |
| num_http_files=num_http_files, | |
| wheels_cache_location=wheels_cache_location, | |
| package_count=num_packages, | |
| wheels_cache_size=wheels_cache_size, | |
| ) | |
| .strip() | |
| ) | |
| logger.info(message) | |
| def list_cache_items(self, options: Values, args: 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 options.list_format == "human": | |
| self.format_for_human(files) | |
| else: | |
| self.format_for_abspath(files) | |
| def format_for_human(self, files: List[str]) -> None: | |
| if not files: | |
| logger.info("No locally built wheels cached.") | |
| return | |
| results = [] | |
| for filename in files: | |
| wheel = os.path.basename(filename) | |
| size = filesystem.format_file_size(filename) | |
| results.append(f" - {wheel} ({size})") | |
| logger.info("Cache contents:\n") | |
| logger.info("\n".join(sorted(results))) | |
| def format_for_abspath(self, files: List[str]) -> None: | |
| if not files: | |
| return | |
| results = [] | |
| for filename in files: | |
| results.append(filename) | |
| logger.info("\n".join(sorted(results))) | |
| def remove_cache_items(self, options: Values, args: 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]) | |
| no_matching_msg = "No matching packages" | |
| if args[0] == "*": | |
| # Only fetch http files if no specific pattern given | |
| files += self._find_http_files(options) | |
| else: | |
| # Add the pattern to the log message | |
| no_matching_msg += ' for pattern "{}"'.format(args[0]) | |
| if not files: | |
| logger.warning(no_matching_msg) | |
| for filename in files: | |
| os.unlink(filename) | |
| logger.verbose("Removed %s", filename) | |
| logger.info("Files removed: %s", len(files)) | |
| def purge_cache(self, options: Values, args: List[Any]) -> None: | |
| if args: | |
| raise CommandError("Too many arguments") | |
| return self.remove_cache_items(options, ["*"]) | |
| def _cache_dir(self, options: Values, subdir: str) -> str: | |
| return os.path.join(options.cache_dir, subdir) | |
| def _find_http_files(self, options: Values) -> List[str]: | |
| http_dir = self._cache_dir(options, "http") | |
| return filesystem.find_files(http_dir, "*") | |
| def _find_wheels(self, options: Values, pattern: str) -> List[str]: | |
| wheel_dir = self._cache_dir(options, "wheels") | |
| # 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) | |