65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
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 head(self, key: str, size: int = 16) -> bytes:
|
|
with self._path(key).open("rb") as file:
|
|
return file.read(size)
|
|
|
|
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))
|