trove/trovedb/trovedb.py

"""
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, db: Sqlite3Trove, object_id: int):
        self._db = 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._db, new_id)

        # Update our node
        self.link(name, tree)
        return tree

    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

    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._db, note_id)
        if ot == "tree":
            return TreeNoteImpl(self._db, 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._db, obj_id)

    def get_root(self) -> TreeNote:
        """Return the root TreeNote (always id=NODE_ROOT_ID)."""
        return TreeNoteImpl(self._db, NODE_ROOT_ID)

def open_trove(path: str | Path, create: bool = False) -> Trove:
    return TroveImpl.open(path, create=create)