159 lines
4.9 KiB
Python
159 lines
4.9 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) 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)
|