Clear cache entries. Args: target: Specific target to clear, or None to clear all Returns: Number of entries cleared
(self, target: Optional[str] = None)
| 177 | logger.error(f"Error writing cache file {cache_path}: {e}") |
| 178 | |
| 179 | def clear(self, target: str | None = None) -> int: |
| 180 | """Clear cache entries. |
| 181 | |
| 182 | Args: |
| 183 | target: Specific target to clear, or None to clear all |
| 184 | |
| 185 | Returns: |
| 186 | Number of entries cleared |
| 187 | """ |
| 188 | cleared_checksums: set[str] = set() |
| 189 | |
| 190 | # Clear memory cache |
| 191 | if target: |
| 192 | keys_to_delete = [ |
| 193 | k for k, v in self._memory_cache.items() |
| 194 | if v.target == target |
| 195 | ] |
| 196 | for k in keys_to_delete: |
| 197 | del self._memory_cache[k] |
| 198 | cleared_checksums.add(k) |
| 199 | else: |
| 200 | cleared_checksums.update(self._memory_cache.keys()) |
| 201 | self._memory_cache.clear() |
| 202 | |
| 203 | # Clear disk cache |
| 204 | try: |
| 205 | for cache_file in self.cache_dir.glob("*.json"): |
| 206 | if target: |
| 207 | with open(cache_file) as f: |
| 208 | data = json.load(f) |
| 209 | if data.get("target") == target: |
| 210 | cache_file.unlink() |
| 211 | cleared_checksums.add(cache_file.stem) |
| 212 | else: |
| 213 | cache_file.unlink() |
| 214 | cleared_checksums.add(cache_file.stem) |
| 215 | except Exception as e: |
| 216 | logger.error(f"Error clearing disk cache: {e}") |
| 217 | |
| 218 | cleared = len(cleared_checksums) |
| 219 | logger.info(f"Cleared {cleared} cache entries for target={target or 'all'}") |
| 220 | return cleared |
| 221 | |
| 222 | def cleanup_expired(self) -> int: |
| 223 | """Remove expired cache entries. |