| from __future__ import annotations | |
| import json | |
| import os | |
| from pathlib import Path | |
| from typing import Any, Dict, Iterable, Optional | |
| def _default_root() -> Path: | |
| return Path(os.getenv("SHIELD_STORAGE_PATH", "/shield")) | |
| class JsonStorage: | |
| def __init__(self, root: Path | None = None) -> None: | |
| self.root = root or _default_root() | |
| def path(self, parts: Iterable[str]) -> Path: | |
| safe_parts = [str(part).strip("/\\") for part in parts if str(part).strip("/\\")] | |
| return self.root.joinpath(*safe_parts) | |
| def read_json(self, *parts: str) -> Optional[Dict[str, Any]]: | |
| path = self.path(parts) | |
| try: | |
| with path.open("r", encoding="utf-8") as handle: | |
| data = json.load(handle) | |
| return data if isinstance(data, dict) else None | |
| except FileNotFoundError: | |
| return None | |
| except Exception: | |
| return None | |
| def write_json(self, data: Dict[str, Any], *parts: str) -> bool: | |
| path = self.path(parts) | |
| try: | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| tmp_path = path.with_suffix(path.suffix + ".tmp") | |
| with tmp_path.open("w", encoding="utf-8") as handle: | |
| json.dump(data, handle, separators=(",", ":"), sort_keys=True) | |
| tmp_path.replace(path) | |
| return True | |
| except Exception: | |
| return False | |
Xet Storage Details
- Size:
- 1.41 kB
- Xet hash:
- 351fe37d34732b0ba8b117579fd24866c3eab46c65d976c8b8c899fad1d10535
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.