Improve database schema for notes-with-children

trovedb/trovedb.py

@@ -2,16 +2,14 @@
 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.
+Depends on db.py (Sqlite3Trove) for storage.
 """
 
-from typing import Optional, Self
+from typing import Optional
 from pathlib import Path
 import datetime as dt
 
 from .db import Sqlite3Trove, NOTE_ROOT_ID
-from .tree import Tree as TreeData
 
 from . import trove as tr
 
@@ -19,7 +17,7 @@ from .trove import Note, Trove, TreeNote, BlobNote, TreeEntry, NoteNotFound, Obj
 
 
 class NoteImpl(Note):
-    """Concrete not implementation"""
+    """Concrete note implementation."""
 
     def __init__(self, parent: 'TroveImpl', object_id: ObjectId):
         self._parent = parent
@@ -37,13 +35,14 @@ class NoteImpl(Note):
 
     @property
     def mtime(self) -> dt.datetime:
-        """Return modification time as Unix timestamp, or None if not set."""
+        """Return modification time as UTC datetime."""
         return self._db.get_mtime(self._object_id)
 
     @property
     def mime(self) -> str:
-        """Return MIME type, defaulting to generic binary stream."""
-        return "application/octet-stream"
+        """Return MIME type from the objects table."""
+        info = self._db.get_info(self._object_id)
+        return info.type if info else "application/octet-stream"
 
     def get_raw_metadata(self, key: str) -> Optional[bytes]:
         return self._db.read_metadata(self._object_id, key)
@@ -61,43 +60,25 @@ class BlobNoteImpl(NoteImpl, BlobNote):
         return data if data is not None else b""
 
     def write(self, data: bytes) -> None:
-        self._db.write_blob(data, self._object_id)
+        self._db.write_content(self._object_id, data)
 
 
 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)
+    """Concrete TreeNote: a tree object backed by the tree_entries table."""
 
     # 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)
+        """Link name to an existing note."""
+        self._db.link(self._object_id, name, note.object_id)
 
     def unlink(self, name: str) -> None:
-        """Remove an entry by name. Raises KeyError if not found."""
-        try:
-            tree = self._read_tree()
-            tree.rm_entry(name)
-            self._flush_tree(tree)
-        except KeyError:
-            pass
+        """Remove an entry by name."""
+        self._db.unlink(self._object_id, name)
 
     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())
+        new_id = self._db.write_tree(b"")
         tree = TreeNoteImpl(self._parent, new_id)
-
-        # Update our node
         self.link(name, tree)
         return tree
 
@@ -107,8 +88,7 @@ class TreeNoteImpl(NoteImpl, TreeNote):
 
     def child(self, name: str) -> Note:
         """Retrieve a child note by name."""
-        tree = self._read_tree()
-        entries = tree.list()
+        entries = self._db.list_tree(self._object_id)
         if name not in entries:
             raise KeyError(f"Entry '{name}' not found")
         child_id = entries[name]
@@ -119,13 +99,12 @@ class TreeNoteImpl(NoteImpl, TreeNote):
 
     def entries(self):
         """Return all entries as an iterable of TreeEntry."""
-        tree = self._read_tree()
-        for name, object_id in tree.list().items():
+        for name, object_id in self._db.list_tree(self._object_id).items():
             yield TreeEntry(name, object_id)
 
     def list(self) -> dict[str, ObjectId]:
         """Return all entries as {name: object_id}."""
-        return self._read_tree().list()
+        return self._db.list_tree(self._object_id)
 
 
 # ---------------------------------------------------------------------------
@@ -145,14 +124,7 @@ class TroveImpl:
     @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 = ?", (NOTE_ROOT_ID,)
-            )
-            db._con.commit()
-        return trove
+        return cls(db)
 
     @property
     def db(self) -> Sqlite3Trove:
@@ -170,23 +142,23 @@ class TroveImpl:
     # Trove protocol
     def get_raw_note(self, note_id: ObjectId) -> 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:
+        info = self._db.get_info(note_id)
+        if info is None:
             raise NoteNotFound(note_id)
-        if ot == "blob":
-            return BlobNoteImpl(self, note_id)
-        if ot == "tree":
+        if self._db.is_tree(note_id) or info.type == "inode/directory":
             return TreeNoteImpl(self, note_id)
-        raise ValueError(f"Unknown object type '{ot}' for id {note_id}")
+        return BlobNoteImpl(self, note_id)
 
-    def create_blob(self, data: bytes | None = None) -> BlobNote:
+    def create_blob(self, data: bytes | None = None,
+                    dtype: str = "application/octet-stream") -> BlobNote:
         """Create a new blob object and return a BlobNote for it."""
-        obj_id = self._db.write_blob(data or b"")
+        obj_id = self._db.write_blob(data or b"", dtype=dtype)
         return BlobNoteImpl(self, obj_id)
 
     def get_root(self) -> TreeNote:
-        """Return the root TreeNote (always id=NODE_ROOT_ID)."""
+        """Return the root TreeNote (always id=NOTE_ROOT_ID)."""
         return TreeNoteImpl(self, NOTE_ROOT_ID)
 
+
 def open_db_trove(path: str | Path, create: bool = False, **kwargs: tr.OpenArguments) -> Trove:
     return TroveImpl.open(path, create=create)