| """HTTP cache implementation. |
| """ |
|
|
| import os |
| from contextlib import contextmanager |
| from typing import Generator, Optional |
|
|
| from pip._vendor.cachecontrol.cache import BaseCache |
| from pip._vendor.cachecontrol.caches import FileCache |
| from pip._vendor.requests.models import Response |
|
|
| from pip._internal.utils.filesystem import adjacent_tmp_file, replace |
| from pip._internal.utils.misc import ensure_dir |
|
|
|
|
| def is_from_cache(response: Response) -> bool: |
| return getattr(response, "from_cache", False) |
|
|
|
|
| @contextmanager |
| def suppressed_cache_errors() -> Generator[None, None, None]: |
| """If we can't access the cache then we can just skip caching and process |
| requests as if caching wasn't enabled. |
| """ |
| try: |
| yield |
| except OSError: |
| pass |
|
|
|
|
| class SafeFileCache(BaseCache): |
| """ |
| A file based cache which is safe to use even when the target directory may |
| not be accessible or writable. |
| """ |
|
|
| def __init__(self, directory: str) -> None: |
| assert directory is not None, "Cache directory must not be None." |
| super().__init__() |
| self.directory = directory |
|
|
| def _get_cache_path(self, name: str) -> str: |
| |
| |
| |
| hashed = FileCache.encode(name) |
| parts = list(hashed[:5]) + [hashed] |
| return os.path.join(self.directory, *parts) |
|
|
| def get(self, key: str) -> Optional[bytes]: |
| path = self._get_cache_path(key) |
| with suppressed_cache_errors(): |
| with open(path, "rb") as f: |
| return f.read() |
|
|
| def set(self, key: str, value: bytes, expires: Optional[int] = None) -> None: |
| path = self._get_cache_path(key) |
| with suppressed_cache_errors(): |
| ensure_dir(os.path.dirname(path)) |
|
|
| with adjacent_tmp_file(path) as f: |
| f.write(value) |
|
|
| replace(f.name, path) |
|
|
| def delete(self, key: str) -> None: |
| path = self._get_cache_path(key) |
| with suppressed_cache_errors(): |
| os.remove(path) |
|
|
