src/black/cache.py

   1 """Caching of formatted files with feature-based invalidation."""
   2
   3 import os
   4 import pickle
   5 from pathlib import Path
   6 import tempfile
   7 from typing import Dict, Iterable, Set, Tuple
   8
   9 from platformdirs import user_cache_dir
  10
  11 from black.mode import Mode
  12
  13 from _black_version import version as __version__
  14
  15
  16 # types
  17 Timestamp = float
  18 FileSize = int
  19 CacheInfo = Tuple[Timestamp, FileSize]
  20 Cache = Dict[str, CacheInfo]
  21
  22
  23 CACHE_DIR = Path(user_cache_dir("black", version=__version__))
  24
  25
  26 def read_cache(mode: Mode) -> Cache:
  27     """Read the cache if it exists and is well formed.
  28
  29     If it is not well formed, the call to write_cache later should resolve the issue.
  30     """
  31     cache_file = get_cache_file(mode)
  32     if not cache_file.exists():
  33         return {}
  34
  35     with cache_file.open("rb") as fobj:
  36         try:
  37             cache: Cache = pickle.load(fobj)
  38         except (pickle.UnpicklingError, ValueError):
  39             return {}
  40
  41     return cache
  42
  43
  44 def get_cache_file(mode: Mode) -> Path:
  45     return CACHE_DIR / f"cache.{mode.get_cache_key()}.pickle"
  46
  47
  48 def get_cache_info(path: Path) -> CacheInfo:
  49     """Return the information used to check if a file is already formatted or not."""
  50     stat = path.stat()
  51     return stat.st_mtime, stat.st_size
  52
  53
  54 def filter_cached(cache: Cache, sources: Iterable[Path]) -> Tuple[Set[Path], Set[Path]]:
  55     """Split an iterable of paths in `sources` into two sets.
  56
  57     The first contains paths of files that modified on disk or are not in the
  58     cache. The other contains paths to non-modified files.
  59     """
  60     todo, done = set(), set()
  61     for src in sources:
  62         res_src = src.resolve()
  63         if cache.get(str(res_src)) != get_cache_info(res_src):
  64             todo.add(src)
  65         else:
  66             done.add(src)
  67     return todo, done
  68
  69
  70 def write_cache(cache: Cache, sources: Iterable[Path], mode: Mode) -> None:
  71     """Update the cache file."""
  72     cache_file = get_cache_file(mode)
  73     try:
  74         CACHE_DIR.mkdir(parents=True, exist_ok=True)
  75         new_cache = {
  76             **cache,
  77             **{str(src.resolve()): get_cache_info(src) for src in sources},
  78         }
  79         with tempfile.NamedTemporaryFile(dir=str(cache_file.parent), delete=False) as f:
  80             pickle.dump(new_cache, f, protocol=4)
  81         os.replace(f.name, cache_file)
  82     except OSError:
  83         pass