(cache_path: str, cache_limit: int, verbose: bool)
| 893 | |
| 894 | |
| 895 | def clean_cache(cache_path: str, cache_limit: int, verbose: bool) -> None: |
| 896 | if not cache_limit: |
| 897 | return |
| 898 | |
| 899 | files = glob.glob(os.path.join(cache_path, "*", "*")) |
| 900 | if not files: |
| 901 | return |
| 902 | |
| 903 | # Store files in list of (filename, size, atime). |
| 904 | stats = [] |
| 905 | for file in files: |
| 906 | try: |
| 907 | stats.append((file, *os.stat(file)[6:8])) |
| 908 | except OSError: |
| 909 | print_error(f'Failed to access cache file "{file}"; skipping.') |
| 910 | |
| 911 | # Sort by most recent access (most sensible to keep) first. Search for the first entry where |
| 912 | # the cache limit is reached. |
| 913 | stats.sort(key=lambda x: x[2], reverse=True) |
| 914 | sum = 0 |
| 915 | for index, stat in enumerate(stats): |
| 916 | sum += stat[1] |
| 917 | if sum > cache_limit: |
| 918 | purge = [x[0] for x in stats[index:]] |
| 919 | count = len(purge) |
| 920 | for file in purge: |
| 921 | try: |
| 922 | os.remove(file) |
| 923 | except OSError: |
| 924 | print_error(f'Failed to remove cache file "{file}"; skipping.') |
| 925 | count -= 1 |
| 926 | if verbose and count: |
| 927 | print_info(f"Purged {count} file{'s' if count else ''} from cache.") |
| 928 | break |
| 929 | |
| 930 | |
| 931 | def prepare_cache(env) -> None: |
nothing calls this directly
no test coverage detected