Evicts oldest cache files (LRU) until cache is under the size limit.
(new_entry: str | None = None)
| 25 | |
| 26 | |
| 27 | def prune_cache(new_entry: str | None = None) -> None: |
| 28 | """Evicts oldest cache files (LRU) until cache is under the size limit.""" |
| 29 | # we use a manifest to avoid tons of os.stat syscalls (slow) |
| 30 | manifest = {} |
| 31 | manifest_path = Paths.download_cache_root() + "manifest.txt" |
| 32 | if os.path.exists(manifest_path): |
| 33 | with open(manifest_path) as f: |
| 34 | manifest = {parts[0]: int(parts[1]) for line in f if (parts := line.strip().split()) and len(parts) == 2} |
| 35 | |
| 36 | if new_entry: |
| 37 | manifest[new_entry] = int(time.time()) # noqa: TID251 |
| 38 | |
| 39 | # evict the least recently used files until under limit |
| 40 | sorted_items = sorted(manifest.items(), key=lambda x: x[1]) |
| 41 | while len(manifest) * CHUNK_SIZE > CACHE_SIZE and sorted_items: |
| 42 | key, _ = sorted_items.pop(0) |
| 43 | try: |
| 44 | os.remove(Paths.download_cache_root() + key) |
| 45 | except OSError: |
| 46 | pass |
| 47 | manifest.pop(key, None) |
| 48 | |
| 49 | # write out manifest |
| 50 | with atomic_write(manifest_path, mode="w", overwrite=True) as f: |
| 51 | f.write('\n'.join(f"{k} {v}" for k, v in manifest.items())) |
| 52 | |
| 53 | class URLFileException(Exception): |
| 54 | pass |