| 29 | |
| 30 | |
| 31 | class FileCache: |
| 32 | def __init__(self, base: Path) -> None: |
| 33 | self.base = base |
| 34 | self.base.mkdir(parents=True, exist_ok=True) |
| 35 | |
| 36 | def _path_for(self, key: str) -> Path: |
| 37 | digest = hash_key(key) |
| 38 | return self.base / digest[:2] / f"{digest}.json" |
| 39 | |
| 40 | def load(self, key: str, *, ttl: int) -> Optional[Dict]: |
| 41 | path = self._path_for(key) |
| 42 | if not path.exists(): |
| 43 | return None |
| 44 | try: |
| 45 | with path.open("r", encoding="utf-8") as handle: |
| 46 | data = json.load(handle) |
| 47 | except (OSError, json.JSONDecodeError): |
| 48 | return None |
| 49 | expires_at = data.get("expires_at") |
| 50 | if expires_at: |
| 51 | try: |
| 52 | expires_ts = time.mktime(time.strptime(expires_at, "%Y-%m-%dT%H:%M:%S")) |
| 53 | if time.time() > expires_ts: |
| 54 | return None |
| 55 | except Exception: |
| 56 | return None |
| 57 | return data.get("payload") |
| 58 | |
| 59 | def save(self, key: str, payload: Dict, *, ttl: int) -> None: |
| 60 | path = self._path_for(key) |
| 61 | path.parent.mkdir(parents=True, exist_ok=True) |
| 62 | data = { |
| 63 | "fetched_at": isoformat(), |
| 64 | "expires_at": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(time.time() + ttl)), |
| 65 | "payload": payload, |
| 66 | } |
| 67 | with path.open("w", encoding="utf-8") as handle: |
| 68 | json.dump(data, handle, ensure_ascii=False, indent=2) |
| 69 | |
| 70 | |
| 71 | @dataclass |