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

1"""HTTP cache implementation. 

2""" 

3 

4import os 

5from contextlib import contextmanager 

6from datetime import datetime 

7from typing import BinaryIO, Generator, Optional, Union 

8 

9from pip._vendor.cachecontrol.cache import SeparateBodyBaseCache 

10from pip._vendor.cachecontrol.caches import SeparateBodyFileCache 

11from pip._vendor.requests.models import Response 

12 

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

14from pip._internal.utils.misc import ensure_dir 

15 

16 

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

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

19 

20 

21@contextmanager 

22def suppressed_cache_errors() -> Generator[None, None, None]: 

23 """If we can't access the cache then we can just skip caching and process 

24 requests as if caching wasn't enabled. 

25 """ 

26 try: 

27 yield 

28 except OSError: 

29 pass 

30 

31 

32class SafeFileCache(SeparateBodyBaseCache): 

33 """ 

34 A file based cache which is safe to use even when the target directory may 

35 not be accessible or writable. 

36 

37 There is a race condition when two processes try to write and/or read the 

38 same entry at the same time, since each entry consists of two separate 

39 files (https://github.com/psf/cachecontrol/issues/324). We therefore have 

40 additional logic that makes sure that both files to be present before 

41 returning an entry; this fixes the read side of the race condition. 

42 

43 For the write side, we assume that the server will only ever return the 

44 same data for the same URL, which ought to be the case for files pip is 

45 downloading. PyPI does not have a mechanism to swap out a wheel for 

46 another wheel, for example. If this assumption is not true, the 

47 CacheControl issue will need to be fixed. 

48 """ 

49 

50 def __init__(self, directory: str) -> None: 

51 assert directory is not None, "Cache directory must not be None." 

52 super().__init__() 

53 self.directory = directory 

54 

55 def _get_cache_path(self, name: str) -> str: 

56 # From cachecontrol.caches.file_cache.FileCache._fn, brought into our 

57 # class for backwards-compatibility and to avoid using a non-public 

58 # method. 

59 hashed = SeparateBodyFileCache.encode(name) 

60 parts = list(hashed[:5]) + [hashed] 

61 return os.path.join(self.directory, *parts) 

62 

63 def get(self, key: str) -> Optional[bytes]: 

64 # The cache entry is only valid if both metadata and body exist. 

65 metadata_path = self._get_cache_path(key) 

66 body_path = metadata_path + ".body" 

67 if not (os.path.exists(metadata_path) and os.path.exists(body_path)): 

68 return None 

69 with suppressed_cache_errors(): 

70 with open(metadata_path, "rb") as f: 

71 return f.read() 

72 

73 def _write(self, path: str, data: bytes) -> None: 

74 with suppressed_cache_errors(): 

75 ensure_dir(os.path.dirname(path)) 

76 

77 with adjacent_tmp_file(path) as f: 

78 f.write(data) 

79 

80 replace(f.name, path) 

81 

82 def set( 

83 self, key: str, value: bytes, expires: Union[int, datetime, None] = None 

84 ) -> None: 

85 path = self._get_cache_path(key) 

86 self._write(path, value) 

87 

88 def delete(self, key: str) -> None: 

89 path = self._get_cache_path(key) 

90 with suppressed_cache_errors(): 

91 os.remove(path) 

92 with suppressed_cache_errors(): 

93 os.remove(path + ".body") 

94 

95 def get_body(self, key: str) -> Optional[BinaryIO]: 

96 # The cache entry is only valid if both metadata and body exist. 

97 metadata_path = self._get_cache_path(key) 

98 body_path = metadata_path + ".body" 

99 if not (os.path.exists(metadata_path) and os.path.exists(body_path)): 

100 return None 

101 with suppressed_cache_errors(): 

102 return open(body_path, "rb") 

103 

104 def set_body(self, key: str, body: bytes) -> None: 

105 path = self._get_cache_path(key) + ".body" 

106 self._write(path, body)