(cls, *, url, fd: Optional[BinaryIO]=None, cache_id=None, tags=None, session=None)
| 231 | |
| 232 | @classmethod |
| 233 | def proxy(cls, *, url, fd: Optional[BinaryIO]=None, cache_id=None, tags=None, session=None): |
| 234 | if cls.get_location() is None: |
| 235 | if fd is None: |
| 236 | logger.warning("FD is set to None but cache is disabled, URL caching has zero effect") |
| 237 | return |
| 238 | |
| 239 | return utils.download_file(url, fd=fd) |
| 240 | |
| 241 | cache_obj = cls(url=url, cache_id=cache_id, tags=tags) |
| 242 | |
| 243 | if cache_obj.is_valid: |
| 244 | logger.debug(f"Loading {cache_obj.cid} from cache") |
| 245 | if fd: |
| 246 | cache_obj.fetch(fd) |
| 247 | return |
| 248 | |
| 249 | try: |
| 250 | cache_obj.download() |
| 251 | cache_obj.save_metadata() |
| 252 | if fd: |
| 253 | cache_obj.fetch(fd) |
| 254 | except Exception: |
| 255 | cache_obj.delete() |
| 256 | raise |
| 257 | |
| 258 | def fetch(self, fd): |
| 259 | with self.cache_file_location.open("rb") as cfd: |
nothing calls this directly
no test coverage detected