import hashlib from pathlib import Path TRASH_DIR = ".trash" class ContentAddressedStorage: def __init__(self, root: str | Path, shard_depth: int = 2) -> None: self._root = Path(root) self._shard_depth = shard_depth def _path(self, key: str) -> Path: shards = [key[i * 2 : i * 2 + 2] for i in range(self._shard_depth)] return self._root.joinpath(*shards, key) def put(self, data: bytes) -> str: key = hashlib.sha256(data).hexdigest() path = self._path(key) if not path.exists(): path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) return key def get(self, key: str) -> bytes: return self._path(key).read_bytes() def exists(self, key: str) -> bool: return self._path(key).exists() def url(self, key: str) -> str: return str(self._path(key)) def _trash_path(self, key: str) -> Path: return self._root / TRASH_DIR / key def trash(self, key: str) -> None: target = self._trash_path(key) target.parent.mkdir(parents=True, exist_ok=True) self._path(key).replace(target) def restore(self, key: str) -> bool: source = self._trash_path(key) target = self._path(key) if not source.exists(): return target.exists() if target.exists(): source.unlink() else: target.parent.mkdir(parents=True, exist_ok=True) source.replace(target) return True def purge(self, key: str) -> None: self._trash_path(key).unlink(missing_ok=True) def trashed(self, key: str) -> bool: return self._trash_path(key).is_file() def trash_url(self, key: str) -> str: return str(self._trash_path(key))