(self, key: str, *, ttl: int)
| 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) |
no test coverage detected