+def sub_twice(regex: Pattern[str], replacement: str, original: str) -> str:
+ """Replace `regex` with `replacement` twice on `original`.
+
+ This is used by string normalization to perform replaces on
+ overlapping matches.
+ """
+ return regex.sub(replacement, regex.sub(replacement, original))
+
+
+CACHE_DIR = Path(user_cache_dir("black", version=__version__))
+
+
+def get_cache_file(line_length: int) -> Path:
+ return CACHE_DIR / f"cache.{line_length}.pickle"
+
+
+def read_cache(line_length: int) -> Cache:
+ """Read the cache if it exists and is well formed.
+
+ If it is not well formed, the call to write_cache later should resolve the issue.
+ """
+ cache_file = get_cache_file(line_length)
+ if not cache_file.exists():
+ return {}
+
+ with cache_file.open("rb") as fobj:
+ try:
+ cache: Cache = pickle.load(fobj)
+ except pickle.UnpicklingError:
+ return {}
+
+ return cache
+
+
+def get_cache_info(path: Path) -> CacheInfo:
+ """Return the information used to check if a file is already formatted or not."""
+ stat = path.stat()
+ return stat.st_mtime, stat.st_size
+
+
+def filter_cached(
+ cache: Cache, sources: Iterable[Path]
+) -> Tuple[List[Path], List[Path]]:
+ """Split a list of paths into two.
+
+ The first list contains paths of files that modified on disk or are not in the
+ cache. The other list contains paths to non-modified files.
+ """
+ todo, done = [], []
+ for src in sources:
+ src = src.resolve()
+ if cache.get(src) != get_cache_info(src):
+ todo.append(src)
+ else:
+ done.append(src)
+ return todo, done
+
+
+def write_cache(cache: Cache, sources: List[Path], line_length: int) -> None:
+ """Update the cache file."""
+ cache_file = get_cache_file(line_length)
+ try:
+ if not CACHE_DIR.exists():
+ CACHE_DIR.mkdir(parents=True)
+ new_cache = {**cache, **{src.resolve(): get_cache_info(src) for src in sources}}
+ with cache_file.open("wb") as fobj:
+ pickle.dump(new_cache, fobj, protocol=pickle.HIGHEST_PROTOCOL)
+ except OSError:
+ pass
+
+
projects / etc / vim.git / blobdiff