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) for storage.
"""

from typing import Optional
from pathlib import Path
import datetime as dt

from .db import Sqlite3Trove, NOTE_ROOT_ID

from . import trove as tr

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


class NoteImpl(Note):
    """Concrete note implementation."""

    def __init__(self, parent: 'TroveImpl', object_id: ObjectId):
        self._parent = parent
        self._db = parent.db
        self._object_id = object_id

    # Note protocol
    @property
    def object_id(self) -> ObjectId:
        return self._object_id

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

    @property
    def mtime(self) -> dt.datetime:
        """Return modification time as UTC datetime."""
        return self._db.get_mtime(self._object_id)

    @property
    def mime(self) -> str:
        """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)

    def set_raw_metadata(self, key: str, value: bytes) -> None:
        self._db.write_metadata(self._object_id, key, value)

    def read_content(self) -> bytes:
        data = self._db.read_object(self._object_id)
        return data if data is not None else b""

    def write_content(self, data: bytes) -> None:
        self._db.write_content(self._object_id, data)


class TreeNoteImpl(NoteImpl, TreeNote):
    """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."""
        self._db.link(self._object_id, name, note.object_id)

    def unlink(self, name: str) -> None:
        """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."""
        new_id = self._db.write_tree(b"")
        tree = TreeNoteImpl(self._parent, new_id)
        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."""
        entries = self._db.list_tree(self._object_id)
        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."""
        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._db.list_tree(self._object_id)


# ---------------------------------------------------------------------------
# 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)
        return cls(db)

    @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: ObjectId) -> Note:
        """Return a BlobNote or TreeNote for the given id, or None if not found."""
        info = self._db.get_info(note_id)
        if info is None:
            raise NoteNotFound(note_id)
        if self._db.is_tree(note_id) or info.type == "inode/directory":
            return TreeNoteImpl(self, note_id)
        return NoteImpl(self, note_id)

    def create_blob(self, data: bytes | None = None,
                    dtype: str = "application/octet-stream") -> Note:
        """Create a new blob object and return a BlobNote for it."""
        obj_id = self._db.write_blob(data or b"", dtype=dtype)
        return NoteImpl(self, obj_id)

    def get_root(self) -> TreeNote:
        """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)