Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pip/_internal/network/cache.py: 41%

1"""HTTP cache implementation. 

2""" 

3 

4import os 

5from contextlib import contextmanager 

6from typing import Generator, Optional 

7 

8from pip._vendor.cachecontrol.cache import BaseCache 

9from pip._vendor.cachecontrol.caches import FileCache 

10from pip._vendor.requests.models import Response 

11 

12from pip._internal.utils.filesystem import adjacent_tmp_file, replace 

13from pip._internal.utils.misc import ensure_dir 

14 

15 

16def is_from_cache(response: Response) -> bool: 

17 return getattr(response, "from_cache", False) 

18 

19 

20@contextmanager 

21def 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 

31class 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)