| 267 | |
| 268 | |
| 269 | class MirrorJSON(Cache): |
| 270 | prefix = "mirrorjson_" |
| 271 | |
| 272 | def __init__(self, src: Path, cache_id=None, tags: Optional[List[str]]=None): |
| 273 | self.src = src |
| 274 | self.tags = tags or [] |
| 275 | self.cid = cache_id or self.cache_id(src) |
| 276 | |
| 277 | @classmethod |
| 278 | def cache_id(cls, src: Path) -> str: |
| 279 | return src.name |
| 280 | |
| 281 | @classmethod |
| 282 | def proxy(cls, *, src: Path) -> Path: |
| 283 | if cls.get_location() is None: # Caching is disabled |
| 284 | return src |
| 285 | |
| 286 | cache_obj = MirrorJSON(src=src) |
| 287 | |
| 288 | if cache_obj.is_valid: |
| 289 | logger.debug(f"Retrieving package mirror JSON {cache_obj.cid} from cache") |
| 290 | return cache_obj.cache_file_location |
| 291 | |
| 292 | # If the mirror is a mounted network drive (common configuration), this would trigger a network traffic/calls |
| 293 | # We want to prevent any network traffic for performance reasons if possible, |
| 294 | # so we check if the path exists AFTER we check for the cache entry, e.g. `cache_obj.is_valid` |
| 295 | if not src.exists(): |
| 296 | return src |
| 297 | |
| 298 | try: |
| 299 | cache_obj.fetch(src=src) |
| 300 | return cache_obj.cache_file_location |
| 301 | except Exception as exc: |
| 302 | cache_obj.delete() |
| 303 | raise exc |
| 304 | |
| 305 | @property |
| 306 | def metadata(self) -> dict: |
| 307 | return { |
| 308 | "src": str(self.src), |
| 309 | "id": self.cid, |
| 310 | "tags": self.tags, |
| 311 | "type": "mirrorjson" |
| 312 | } |
| 313 | |
| 314 | def fetch(self, src: Path): |
| 315 | shutil.copyfile(src=src, dst=self.cache_file_location, follow_symlinks=True) |
| 316 | self.save_metadata() |
| 317 | |
| 318 | |
| 319 | class MirrorFile(Cache): |