Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/black/cache.py: 46%

1"""Caching of formatted files with feature-based invalidation.""" 

2 

3import hashlib 

4import os 

5import pickle 

6import sys 

7import tempfile 

8from collections.abc import Iterable 

9from dataclasses import dataclass, field 

10from pathlib import Path 

11from typing import NamedTuple 

12 

13from platformdirs import user_cache_dir 

14 

15from _black_version import version as __version__ 

16from black.mode import Mode 

17from black.output import err 

18 

19if sys.version_info >= (3, 11): 

20 from typing import Self 

21else: 

22 from typing_extensions import Self 

23 

24 

25class FileData(NamedTuple): 

26 st_mtime: float 

27 st_size: int 

28 hash: str 

29 

30 

31def get_cache_dir(cache_dir: Path | None = None) -> Path: 

32 """Get the cache directory used by black. 

33 

34 Users can customize this directory directly or, on all systems, using the 

35 `BLACK_CACHE_DIR` environment variable. By default, the cache directory is the 

36 user cache directory under the black application. 

37 

38 This result is immediately set to a constant `black.cache.CACHE_DIR` as to avoid 

39 repeated calls. 

40 """ 

41 # NOTE: Function mostly exists as a clean way to test getting the cache directory. 

42 default_cache_dir = user_cache_dir("black") 

43 if cache_dir is None: 

44 cache_dir = Path(os.environ.get("BLACK_CACHE_DIR", default_cache_dir)) 

45 cache_dir = cache_dir / __version__ 

46 return cache_dir 

47 

48 

49CACHE_DIR = get_cache_dir() 

50 

51 

52def get_cache_file(mode: Mode, cache_dir: Path) -> Path: 

53 return cache_dir / f"cache.{mode.get_cache_key()}.pickle" 

54 

55 

56@dataclass 

57class Cache: 

58 mode: Mode 

59 cache_file: Path 

60 file_data: dict[str, FileData] = field(default_factory=dict) 

61 

62 @classmethod 

63 def read(cls, mode: Mode, cache_dir: Path = CACHE_DIR) -> Self: 

64 """Read the cache if it exists and is well-formed. 

65 

66 If it is not well-formed, the call to write later should 

67 resolve the issue. 

68 """ 

69 cache_file = get_cache_file(mode, cache_dir) 

70 try: 

71 exists = cache_file.exists() 

72 except OSError as e: 

73 # Likely file too long; see #4172 and #4174 

74 err(f"Unable to read cache file {cache_file} due to {e}") 

75 return cls(mode, cache_file) 

76 if not exists: 

77 return cls(mode, cache_file) 

78 

79 with cache_file.open("rb") as fobj: 

80 try: 

81 data: dict[str, tuple[float, int, str]] = pickle.load(fobj) 

82 file_data = {k: FileData(*v) for k, v in data.items()} 

83 except ( 

84 pickle.UnpicklingError, 

85 EOFError, 

86 ValueError, 

87 IndexError, 

88 PermissionError, 

89 ): 

90 return cls(mode, cache_file) 

91 

92 return cls(mode, cache_file, file_data) 

93 

94 @staticmethod 

95 def hash_digest(path: Path) -> str: 

96 """Return hash digest for path.""" 

97 

98 data = path.read_bytes() 

99 return hashlib.sha256(data).hexdigest() 

100 

101 @staticmethod 

102 def get_file_data(path: Path) -> FileData: 

103 """Return file data for path.""" 

104 

105 stat = path.stat() 

106 hash = Cache.hash_digest(path) 

107 return FileData(stat.st_mtime, stat.st_size, hash) 

108 

109 def is_changed(self, source: Path) -> bool: 

110 """Check if source has changed compared to cached version.""" 

111 res_src = source.resolve() 

112 old = self.file_data.get(str(res_src)) 

113 if old is None: 

114 return True 

115 

116 st = res_src.stat() 

117 if st.st_size != old.st_size: 

118 return True 

119 if st.st_mtime != old.st_mtime: 

120 new_hash = Cache.hash_digest(res_src) 

121 if new_hash != old.hash: 

122 return True 

123 return False 

124 

125 def filtered_cached(self, sources: Iterable[Path]) -> tuple[set[Path], set[Path]]: 

126 """Split an iterable of paths in `sources` into two sets. 

127 

128 The first contains paths of files that modified on disk or are not in the 

129 cache. The other contains paths to non-modified files. 

130 """ 

131 changed: set[Path] = set() 

132 done: set[Path] = set() 

133 for src in sources: 

134 if self.is_changed(src): 

135 changed.add(src) 

136 else: 

137 done.add(src) 

138 return changed, done 

139 

140 def write(self, sources: Iterable[Path]) -> None: 

141 """Update the cache file data and write a new cache file.""" 

142 self.file_data.update( 

143 **{str(src.resolve()): Cache.get_file_data(src) for src in sources} 

144 ) 

145 try: 

146 self.cache_file.parent.mkdir(parents=True, exist_ok=True) 

147 with tempfile.NamedTemporaryFile( 

148 dir=str(self.cache_file.parent), delete=False 

149 ) as f: 

150 # We store raw tuples in the cache because it's faster. 

151 data: dict[str, tuple[float, int, str]] = { 

152 k: (*v,) for k, v in self.file_data.items() 

153 } 

154 pickle.dump(data, f, protocol=4) 

155 os.replace(f.name, self.cache_file) 

156 except OSError: 

157 pass