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
import datetime as dt

from .db import Sqlite3Trove, NOTE_ROOT_ID
from .tree import Tree as TreeData

from . import trove as tr

from .trove import Note, Trove, TreeNote, BlobNote, TreeEntry, NoteNotFound


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

    @property
    def readonly(self) -> bool:
        return False

    @property
    def mtime(self) -> dt.datetime:
        """Return modification time as Unix timestamp, or None if not set."""
        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"

    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."""
        try:
            tree = self._read_tree()
            tree.rm_entry(name)
            self._flush_tree(tree)
        except KeyError:
            pass

    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]
        value = self._parent.get_raw_note(child_id)
        if value is None:
            raise KeyError(f"Entry '{name}' has no value")
        return value

    def entries(self):
        """Return all entries as an iterable of TreeEntry."""
        tree = self._read_tree()
        for name, object_id in tree.list().items():
            yield TreeEntry(name, object_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 = ?", (NOTE_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) -> 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:
            raise NoteNotFound(note_id)
        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, NOTE_ROOT_ID)

def open_db_trove(path: str | Path, create: bool = False, **kwargs: tr.OpenArguments) -> Trove:
    return TroveImpl.open(path, create=create)