python (3.11.7)

(root)/
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 ESC[4;38;5;81mCacheCommand(ESC[4;38;5;149mCommand):
      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)