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

   1 """HTTP cache implementation.
   2 """
   3
   4 import os
   5 from contextlib import contextmanager
   6 from typing import Generator, Optional
   7
   8 from pip._vendor.cachecontrol.cache import BaseCache
   9 from pip._vendor.cachecontrol.caches import FileCache
  10 from pip._vendor.requests.models import Response
  11
  12 from pip._internal.utils.filesystem import adjacent_tmp_file, replace
  13 from pip._internal.utils.misc import ensure_dir
  14
  15
  16 def is_from_cache(response: Response) -> bool:
  17     return getattr(response, "from_cache", False)
  18
  19
  20 @contextmanager
  21 def suppressed_cache_errors() -> Generator[None, None, None]:
  22     """If we can't access the cache then we can just skip caching and process
  23     requests as if caching wasn't enabled.
  24     """
  25     try:
  26         yield
  27     except OSError:
  28         pass
  29
  30
  31 class SafeFileCache(BaseCache):
  32     """
  33     A file based cache which is safe to use even when the target directory may
  34     not be accessible or writable.
  35     """
  36
  37     def __init__(self, directory: str) -> None:
  38         assert directory is not None, "Cache directory must not be None."
  39         super().__init__()
  40         self.directory = directory
  41
  42     def _get_cache_path(self, name: str) -> str:
  43         # From cachecontrol.caches.file_cache.FileCache._fn, brought into our
  44         # class for backwards-compatibility and to avoid using a non-public
  45         # method.
  46         hashed = FileCache.encode(name)
  47         parts = list(hashed[:5]) + [hashed]
  48         return os.path.join(self.directory, *parts)
  49
  50     def get(self, key: str) -> Optional[bytes]:
  51         path = self._get_cache_path(key)
  52         with suppressed_cache_errors():
  53             with open(path, "rb") as f:
  54                 return f.read()
  55
  56     def set(self, key: str, value: bytes, expires: Optional[int] = None) -> None:
  57         path = self._get_cache_path(key)
  58         with suppressed_cache_errors():
  59             ensure_dir(os.path.dirname(path))
  60
  61             with adjacent_tmp_file(path) as f:
  62                 f.write(value)
  63
  64             replace(f.name, path)
  65
  66     def delete(self, key: str) -> None:
  67         path = self._get_cache_path(key)
  68         with suppressed_cache_errors():
  69             os.remove(path)