162 lines
5 KiB
Python
162 lines
5 KiB
Python
"""
|
|
trovedb.py — Concrete implementation of Trove protocols backed by Sqlite3Trove.
|
|
|
|
Implements BlobNote, TreeNote, and Trove protocols defined in trove.py.
|
|
Depends on db.py (Sqlite3Trove) and tree.py (Tree) for storage and
|
|
tree serialization respectively.
|
|
"""
|
|
|
|
from typing import Optional, Self
|
|
from pathlib import Path
|
|
|
|
from .db import Sqlite3Trove
|
|
from .tree import Tree as TreeData
|
|
from .trove import NODE_ROOT_ID, Note, Trove, TreeNote, BlobNote
|
|
|
|
|
|
class NoteImpl(Note):
|
|
"""Concrete not implementation"""
|
|
|
|
def __init__(self, parent: 'TroveImpl', object_id: int):
|
|
self._parent = parent
|
|
self._db = parent.db
|
|
self._object_id = object_id
|
|
|
|
# Note protocol
|
|
@property
|
|
def object_id(self) -> int:
|
|
return self._object_id
|
|
|
|
def get_raw_metadata(self, key: str) -> Optional[bytes]:
|
|
return self._db.read_metadata(self._object_id, key)
|
|
|
|
def set_raw_metadata(self, key: str, value: bytes) -> None:
|
|
self._db.write_metadata(self._object_id, key, value)
|
|
|
|
|
|
class BlobNoteImpl(NoteImpl, BlobNote):
|
|
"""Concrete BlobNote: a blob object in the store with metadata access."""
|
|
|
|
# Blob protocol
|
|
def read(self) -> bytes:
|
|
data = self._db.read_object(self._object_id)
|
|
return data if data is not None else b""
|
|
|
|
def write(self, data: bytes) -> None:
|
|
self._db.write_blob(data, self._object_id)
|
|
|
|
|
|
class TreeNoteImpl(NoteImpl, TreeNote):
|
|
"""Concrete TreeNote: a tree object in the store with metadata access."""
|
|
|
|
def _read_tree(self) -> TreeData:
|
|
data = self._db.read_object(self._object_id)
|
|
return TreeData(data if data else None)
|
|
|
|
def _flush_tree(self, tree: TreeData) -> None:
|
|
self._db.write_tree(tree.serialize(), self._object_id)
|
|
|
|
# Tree protocol
|
|
def link(self, name: str, note: Note) -> None:
|
|
"""Link name to an existing note (blob or tree)."""
|
|
tree = self._read_tree()
|
|
tree.set_entry(name, note.object_id)
|
|
self._flush_tree(tree)
|
|
|
|
def unlink(self, name: str) -> None:
|
|
"""Remove an entry by name. Raises KeyError if not found."""
|
|
tree = self._read_tree()
|
|
tree.rm_entry(name)
|
|
self._flush_tree(tree)
|
|
|
|
def mkdir(self, name: str) -> 'TreeNoteImpl':
|
|
"""Create a new empty tree, link it under name, and return it."""
|
|
|
|
# Create the new node
|
|
new_id = self._db.write_tree(TreeData().serialize())
|
|
tree = TreeNoteImpl(self._parent, new_id)
|
|
|
|
# Update our node
|
|
self.link(name, tree)
|
|
return tree
|
|
|
|
def rmdir(self, name: str) -> None:
|
|
"""Remove a directory from the tree."""
|
|
self.unlink(name)
|
|
|
|
def child(self, name: str) -> Note:
|
|
"""Retrieve a child note by name."""
|
|
tree = self._read_tree()
|
|
entries = tree.list()
|
|
if name not in entries:
|
|
raise KeyError(f"Entry '{name}' not found")
|
|
child_id = entries[name]
|
|
return self._parent.get_raw_note(child_id)
|
|
|
|
def list(self) -> dict[str, int]:
|
|
"""Return all entries as {name: object_id}."""
|
|
return self._read_tree().list()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Trove
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TroveImpl:
|
|
"""
|
|
Concrete Trove: top-level API backed by a Sqlite3Trove database.
|
|
|
|
Use TroveImpl.open() to get an instance.
|
|
"""
|
|
|
|
def __init__(self, db: Sqlite3Trove):
|
|
self._db = db
|
|
|
|
@classmethod
|
|
def open(cls, path: str | Path, create: bool = False) -> "TroveImpl":
|
|
db = Sqlite3Trove.open(path, create=create)
|
|
trove = cls(db)
|
|
if create:
|
|
# Root was written as a blob by Sqlite3Trove.open(); fix its type.
|
|
db._con.execute(
|
|
"UPDATE objects SET type = 'tree' WHERE id = ?", (NODE_ROOT_ID,)
|
|
)
|
|
db._con.commit()
|
|
return trove
|
|
|
|
@property
|
|
def db(self) -> Sqlite3Trove:
|
|
return self._db
|
|
|
|
def close(self) -> None:
|
|
self._db.close()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_):
|
|
self.close()
|
|
|
|
# Trove protocol
|
|
def get_raw_note(self, note_id: int) -> Optional[Note]:
|
|
"""Return a BlobNote or TreeNote for the given id, or None if not found."""
|
|
ot = self._db.get_object_type(note_id)
|
|
if ot is None:
|
|
return None
|
|
if ot == "blob":
|
|
return BlobNoteImpl(self, note_id)
|
|
if ot == "tree":
|
|
return TreeNoteImpl(self, note_id)
|
|
raise ValueError(f"Unknown object type '{ot}' for id {note_id}")
|
|
|
|
def create_blob(self, data: bytes | None = None) -> BlobNote:
|
|
"""Create a new blob object and return a BlobNote for it."""
|
|
obj_id = self._db.write_blob(data or b"")
|
|
return BlobNoteImpl(self, obj_id)
|
|
|
|
def get_root(self) -> TreeNote:
|
|
"""Return the root TreeNote (always id=NODE_ROOT_ID)."""
|
|
return TreeNoteImpl(self, NODE_ROOT_ID)
|
|
|
|
def open_trove(path: str | Path, create: bool = False) -> Trove:
|
|
return TroveImpl.open(path, create=create)
|