venv/lib/python3.11/site-packages/pip/_internal/commands/cache.py

   1 import os
   2 import textwrap
   3 from optparse import Values
   4 from typing import Any, List
   5
   6 import pip._internal.utils.filesystem as filesystem
   7 from pip._internal.cli.base_command import Command
   8 from pip._internal.cli.status_codes import ERROR, SUCCESS
   9 from pip._internal.exceptions import CommandError, PipError
  10 from pip._internal.utils.logging import getLogger
  11
  12 logger = getLogger(__name__)
  13
  14
  15 class CacheCommand(Command):
  16     """
  17     Inspect and manage pip's wheel cache.
  18
  19     Subcommands:
  20
  21     - dir: Show the cache directory.
  22     - info: Show information about the cache.
  23     - list: List filenames of packages stored in the cache.
  24     - remove: Remove one or more package from the cache.
  25     - purge: Remove all items from the cache.
  26
  27     ``<pattern>`` can be a glob expression or a package name.
  28     """
  29
  30     ignore_require_venv = True
  31     usage = """
  32         %prog dir
  33         %prog info
  34         %prog list [<pattern>] [--format=[human, abspath]]
  35         %prog remove <pattern>
  36         %prog purge
  37     """
  38
  39     def add_options(self) -> None:
  40         self.cmd_opts.add_option(
  41             "--format",
  42             action="store",
  43             dest="list_format",
  44             default="human",
  45             choices=("human", "abspath"),
  46             help="Select the output format among: human (default) or abspath",
  47         )
  48
  49         self.parser.insert_option_group(0, self.cmd_opts)
  50
  51     def run(self, options: Values, args: List[str]) -> int:
  52         handlers = {
  53             "dir": self.get_cache_dir,
  54             "info": self.get_cache_info,
  55             "list": self.list_cache_items,
  56             "remove": self.remove_cache_items,
  57             "purge": self.purge_cache,
  58         }
  59
  60         if not options.cache_dir:
  61             logger.error("pip cache commands can not function since cache is disabled.")
  62             return ERROR
  63
  64         # Determine action
  65         if not args or args[0] not in handlers:
  66             logger.error(
  67                 "Need an action (%s) to perform.",
  68                 ", ".join(sorted(handlers)),
  69             )
  70             return ERROR
  71
  72         action = args[0]
  73
  74         # Error handling happens here, not in the action-handlers.
  75         try:
  76             handlers[action](options, args[1:])
  77         except PipError as e:
  78             logger.error(e.args[0])
  79             return ERROR
  80
  81         return SUCCESS
  82
  83     def get_cache_dir(self, options: Values, args: List[Any]) -> None:
  84         if args:
  85             raise CommandError("Too many arguments")
  86
  87         logger.info(options.cache_dir)
  88
  89     def get_cache_info(self, options: Values, args: List[Any]) -> None:
  90         if args:
  91             raise CommandError("Too many arguments")
  92
  93         num_http_files = len(self._find_http_files(options))
  94         num_packages = len(self._find_wheels(options, "*"))
  95
  96         http_cache_location = self._cache_dir(options, "http")
  97         wheels_cache_location = self._cache_dir(options, "wheels")
  98         http_cache_size = filesystem.format_directory_size(http_cache_location)
  99         wheels_cache_size = filesystem.format_directory_size(wheels_cache_location)
 100
 101         message = (
 102             textwrap.dedent(
 103                 """
 104                     Package index page cache location: {http_cache_location}
 105                     Package index page cache size: {http_cache_size}
 106                     Number of HTTP files: {num_http_files}
 107                     Locally built wheels location: {wheels_cache_location}
 108                     Locally built wheels size: {wheels_cache_size}
 109                     Number of locally built wheels: {package_count}
 110                 """
 111             )
 112             .format(
 113                 http_cache_location=http_cache_location,
 114                 http_cache_size=http_cache_size,
 115                 num_http_files=num_http_files,
 116                 wheels_cache_location=wheels_cache_location,
 117                 package_count=num_packages,
 118                 wheels_cache_size=wheels_cache_size,
 119             )
 120             .strip()
 121         )
 122
 123         logger.info(message)
 124
 125     def list_cache_items(self, options: Values, args: List[Any]) -> None:
 126         if len(args) > 1:
 127             raise CommandError("Too many arguments")
 128
 129         if args:
 130             pattern = args[0]
 131         else:
 132             pattern = "*"
 133
 134         files = self._find_wheels(options, pattern)
 135         if options.list_format == "human":
 136             self.format_for_human(files)
 137         else:
 138             self.format_for_abspath(files)
 139
 140     def format_for_human(self, files: List[str]) -> None:
 141         if not files:
 142             logger.info("No locally built wheels cached.")
 143             return
 144
 145         results = []
 146         for filename in files:
 147             wheel = os.path.basename(filename)
 148             size = filesystem.format_file_size(filename)
 149             results.append(f" - {wheel} ({size})")
 150         logger.info("Cache contents:\n")
 151         logger.info("\n".join(sorted(results)))
 152
 153     def format_for_abspath(self, files: List[str]) -> None:
 154         if not files:
 155             return
 156
 157         results = []
 158         for filename in files:
 159             results.append(filename)
 160
 161         logger.info("\n".join(sorted(results)))
 162
 163     def remove_cache_items(self, options: Values, args: List[Any]) -> None:
 164         if len(args) > 1:
 165             raise CommandError("Too many arguments")
 166
 167         if not args:
 168             raise CommandError("Please provide a pattern")
 169
 170         files = self._find_wheels(options, args[0])
 171
 172         no_matching_msg = "No matching packages"
 173         if args[0] == "*":
 174             # Only fetch http files if no specific pattern given
 175             files += self._find_http_files(options)
 176         else:
 177             # Add the pattern to the log message
 178             no_matching_msg += ' for pattern "{}"'.format(args[0])
 179
 180         if not files:
 181             logger.warning(no_matching_msg)
 182
 183         for filename in files:
 184             os.unlink(filename)
 185             logger.verbose("Removed %s", filename)
 186         logger.info("Files removed: %s", len(files))
 187
 188     def purge_cache(self, options: Values, args: List[Any]) -> None:
 189         if args:
 190             raise CommandError("Too many arguments")
 191
 192         return self.remove_cache_items(options, ["*"])
 193
 194     def _cache_dir(self, options: Values, subdir: str) -> str:
 195         return os.path.join(options.cache_dir, subdir)
 196
 197     def _find_http_files(self, options: Values) -> List[str]:
 198         http_dir = self._cache_dir(options, "http")
 199         return filesystem.find_files(http_dir, "*")
 200
 201     def _find_wheels(self, options: Values, pattern: str) -> List[str]:
 202         wheel_dir = self._cache_dir(options, "wheels")
 203
 204         # The wheel filename format, as specified in PEP 427, is:
 205         #     {distribution}-{version}(-{build})?-{python}-{abi}-{platform}.whl
 206         #
 207         # Additionally, non-alphanumeric values in the distribution are
 208         # normalized to underscores (_), meaning hyphens can never occur
 209         # before `-{version}`.
 210         #
 211         # Given that information:
 212         # - If the pattern we're given contains a hyphen (-), the user is
 213         #   providing at least the version. Thus, we can just append `*.whl`
 214         #   to match the rest of it.
 215         # - If the pattern we're given doesn't contain a hyphen (-), the
 216         #   user is only providing the name. Thus, we append `-*.whl` to
 217         #   match the hyphen before the version, followed by anything else.
 218         #
 219         # PEP 427: https://www.python.org/dev/peps/pep-0427/
 220         pattern = pattern + ("*.whl" if "-" in pattern else "-*.whl")
 221
 222         return filesystem.find_files(wheel_dir, pattern)