Compare commits

..

7 commits

15 changed files with 585 additions and 59 deletions

View file

@ -12,6 +12,7 @@ dependencies = []
[project.scripts]
trove = "trovedb.cli:cli_main"
qtrove = "trovedb.qgui.qtrove:main"
[project.optional-dependencies]
dev = [

View file

@ -9,21 +9,22 @@ similar ideas to git storage.
import argparse
import sqlite3
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path
NOTE_ROOT_ID = 1
NOTE_ROOT_ID = uuid.UUID(int=0)
_SCHEMA = """
CREATE TABLE IF NOT EXISTS objects (
id INTEGER PRIMARY KEY,
id TEXT PRIMARY KEY,
type TEXT NOT NULL CHECK(type IN ('blob', 'tree')),
data BLOB,
modified TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS metadata (
id INTEGER NOT NULL REFERENCES objects(id) ON DELETE CASCADE,
id TEXT NOT NULL REFERENCES objects(id) ON DELETE CASCADE,
key TEXT NOT NULL,
value BLOB,
PRIMARY KEY (id, key)
@ -32,14 +33,21 @@ CREATE TABLE IF NOT EXISTS metadata (
CREATE TABLE IF NOT EXISTS labels
(
label TEXT PRIMARY KEY,
id INTEGER NOT NULL REFERENCES objects(id) ON DELETE CASCADE
id TEXT NOT NULL REFERENCES objects(id) ON DELETE CASCADE
);
"""
type SqlObjectId = str | uuid.UUID
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
def _sql_id(id: SqlObjectId | None) -> str | None:
if id is None:
return None
return (
id if isinstance(id, str) else str(id)
)
def _initialize_db(con: sqlite3.Connection):
con.executescript(_SCHEMA)
@ -71,7 +79,7 @@ class Sqlite3Trove:
con.commit()
obj = cls(con)
if initialize:
obj.write_blob(b"", NODE_ROOT_ID)
obj.write_tree(b"", NOTE_ROOT_ID)
return obj
def close(self):
@ -87,86 +95,88 @@ class Sqlite3Trove:
# CRUD operations
# ------------------------------------------------------------------
def get_object_type(self, object_id: int) -> str | None:
def get_object_type(self, object_id: SqlObjectId) -> str | None:
"""Return the type column for an object, or None if not found."""
row = self._con.execute(
"SELECT type FROM objects WHERE id = ?", (object_id,)
"SELECT type FROM objects WHERE id = ?", (_sql_id(object_id),)
).fetchone()
return row["type"] if row else None
def read_object(self, object_id: int) -> bytes | None:
def read_object(self, object_id: SqlObjectId) -> bytes | None:
"""Return raw data for a blob object, or None if not found."""
row = self._con.execute(
"SELECT data, type FROM objects WHERE id = ?", (object_id,)
"SELECT data, type FROM objects WHERE id = ?", (_sql_id(object_id),)
).fetchone()
if row is None:
return None
return bytes(row["data"]) if row["data"] is not None else b""
def read_metadata(self, object_id: int, key: str) -> bytes | None:
def get_mtime(self, object_id: SqlObjectId) -> datetime | None:
"""Return the modified timestamp for an object, or None if not found."""
row = self._con.execute(
"SELECT modified FROM objects WHERE id = ?", (_sql_id(object_id),)
).fetchone()
if row is None:
return None
return datetime.fromisoformat(row["modified"])
def read_metadata(self, object_id: SqlObjectId, key: str) -> bytes | None:
"""Return raw metadata value for (uuid, key), or None if not found."""
row = self._con.execute(
"SELECT value FROM metadata WHERE id = ? AND key = ?", (object_id, key)
"SELECT value FROM metadata WHERE id = ? AND key = ?", (_sql_id(object_id), key)
).fetchone()
if row is None:
return None
return bytes(row["value"]) if row["value"] is not None else b""
def write_metadata(self, object_id: int, key: str, value: bytes) -> None:
def write_metadata(self, object_id: SqlObjectId, key: str, value: bytes) -> None:
"""Upsert a metadata row. db.py has no write_metadata, so we go direct."""
self._con.execute(
"INSERT OR REPLACE INTO metadata (id, key, value) VALUES (?, ?, ?)",
(object_id, key, value),
(_sql_id(object_id), key, value),
)
self._con.commit()
def _write_object(self, data: bytes, dtype: str, object_id: int | None = None) -> int:
def _write_object(self, data: bytes, dtype: str, object_id: str | uuid.UUID | None = None) -> str:
"""
Insert or replace an object. Returns the id.
If object_id is None, creates a new object with auto-assigned ID.
If object_id is None, creates a new object with a new UUID.
If object_id is provided, updates or creates the object with that ID.
"""
modified = _now()
if object_id is None:
cur = self._con.execute(
"INSERT INTO objects (type, data, modified) VALUES (?, ?, ?)",
(dtype, data, modified)
)
self._con.commit()
assert cur.lastrowid is not None
return cur.lastrowid
else:
self._con.execute(
"INSERT OR REPLACE INTO objects (id, type, data, modified) VALUES (?, ?, ?, ?)",
(object_id, dtype, data, modified)
)
self._con.commit()
return object_id
object_id = uuid.uuid4()
self._con.execute(
"INSERT OR REPLACE INTO objects (id, type, data, modified) VALUES (?, ?, ?, ?)",
(_sql_id(object_id), dtype, data, modified)
)
self._con.commit()
return _sql_id(object_id)
def write_blob(self, data: bytes, object_id: int | None = None) -> int:
def write_blob(self, data: bytes, object_id: SqlObjectId | None = None) -> str:
"""
Insert or replace a blob. Returns the id.
Pass object_id to update an existing object.
"""
return self._write_object(data, "blob", object_id)
return self._write_object(data, "blob", _sql_id(object_id))
def write_tree(self, data: bytes, object_id: int | None = None) -> int:
def write_tree(self, data: bytes, object_id: SqlObjectId | None = None) -> str:
"""Write a tree-typed object. Returns the assigned id."""
return self._write_object(data, "tree", object_id)
return self._write_object(data, "tree", _sql_id(object_id))
def delete_object(self, object_id: int) -> bool:
def delete_object(self, object_id: SqlObjectId) -> bool:
"""
Delete a blob and all its metadata rows.
Returns True if an object was deleted, False if id not found.
Foreign key cascade handles the metadata rows.
"""
cur = self._con.execute(
"DELETE FROM objects WHERE id = ?", (object_id,)
"DELETE FROM objects WHERE id = ?", (_sql_id(object_id),)
)
self._con.commit()
return cur.rowcount > 0
def get_label(self, label: str) -> int | None:
def get_label(self, label: str) -> SqlObjectId | None:
"""
Return the ID associated with a label, or None if not found.
"""
@ -175,16 +185,16 @@ class Sqlite3Trove:
).fetchone()
if row is None:
return None
return row["id"]
return uuid.UUID(row["id"])
def set_label(self, label: str, object_id: int) -> None:
def set_label(self, label: str, object_id: SqlObjectId) -> None:
"""
Set a label to point to an ID. Creates or updates the label.
The ID must exist in the objects table.
"""
self._con.execute(
"INSERT OR REPLACE INTO labels (label, id) VALUES (?, ?)",
(label, object_id),
(label, _sql_id(object_id)),
)
self._con.commit()
@ -198,7 +208,7 @@ class Sqlite3Trove:
self._con.commit()
return cur.rowcount > 0
def list_labels(self) -> list[tuple[str, int]]:
def list_labels(self) -> list[tuple[str, str]]:
"""
Return all labels as a list of (label, id) tuples.
"""
@ -220,21 +230,21 @@ def main():
# Get blob
get_parser = subparsers.add_parser("get", help="Get a blob object by ID")
get_parser.add_argument("id", type=int, help="ID of the blob to retrieve")
get_parser.add_argument("id", help="ID of the blob to retrieve")
# Write blob
write_parser = subparsers.add_parser("write", help="Write data to a blob")
write_parser.add_argument("data", help="Data to write (as string, will be encoded as UTF-8)")
write_parser.add_argument("--id", type=int, help="ID of existing blob to update (optional)")
write_parser.add_argument("--id", help="ID of existing blob to update (optional)")
# Delete blob
delete_parser = subparsers.add_parser("delete", help="Delete a blob by ID")
delete_parser.add_argument("id", type=int, help="ID of the blob to delete")
delete_parser.add_argument("id", help="ID of the blob to delete")
# Set label
setlabel_parser = subparsers.add_parser("setlabel", help="Create or update a label to point to an ID")
setlabel_parser.add_argument("label", help="Label name")
setlabel_parser.add_argument("id", type=int, help="ID to associate with the label")
setlabel_parser.add_argument("id", help="ID to associate with the label")
# Remove label
rmlabel_parser = subparsers.add_parser("rmlabel", help="Delete a label")

View file

@ -1,6 +1,7 @@
import os
import sqlite3
import tempfile
import datetime as dt
from pathlib import Path
from typing import Optional, Dict, List, Self, Iterable
from .trove import Note, Trove, TreeNote, BlobNote, Blob, Tree, BadNoteType, TreeEntry, NoteNotFound
@ -24,6 +25,24 @@ class FSNote(Note):
raise ValueError("Note not yet saved to disk")
return self._inode
@property
def mtime(self):
"""Return modification time as datetime."""
stat = self._path.stat()
return dt.datetime.fromtimestamp(stat.st_mtime, tz=dt.timezone.utc)
@property
def readonly(self) -> bool:
"""Check if the note is readonly based on file permissions."""
if self._inode is None:
return False
return not os.access(self._path, os.W_OK)
@property
def mime(self) -> str:
"""Return MIME type, defaulting to generic binary stream."""
return "application/octet-stream"
@property
def _path(self) -> Path:
if self._fs_path is not None:
@ -60,6 +79,11 @@ class FSBlobNote(FSNote, BlobNote):
pass
class FSTreeNote(FSNote, TreeNote):
@property
def mime(self) -> str:
"""Return MIME type for directory/tree nodes."""
return "inode/directory"
def link(self, name: str, note: Note):
if not isinstance(note, FSBlobNote):
raise BadNoteType("Only blob notes can be linked")

View file

@ -194,20 +194,26 @@ class TroveFuseOps(pyfuse3.Operations):
attr.st_nlink = 1
attr.st_uid = os.getuid()
attr.st_gid = os.getgid()
mtime_ns = int(note.mtime.timestamp() * 1e9)
now_ns = int(time.time() * 1e9)
attr.st_atime_ns = now_ns
attr.st_mtime_ns = now_ns
attr.st_ctime_ns = now_ns
attr.st_mtime_ns = mtime_ns
attr.st_ctime_ns = mtime_ns
attr.generation = 0
attr.entry_timeout = 5.0
attr.attr_timeout = 5.0
# Determine permissions based on readonly property
if is_tree:
attr.st_mode = stat.S_IFDIR | 0o755
mode = 0o755 if not note.readonly else 0o555
attr.st_mode = stat.S_IFDIR | mode
attr.st_size = 0
attr.st_blksize = 512
attr.st_blocks = 0
else:
attr.st_mode = stat.S_IFREG | 0o644
mode = 0o644 if not note.readonly else 0o444
attr.st_mode = stat.S_IFREG | mode
attr.st_size = size
attr.st_blksize = 512
attr.st_blocks = (size + 511) // 512

View file

@ -0,0 +1,77 @@
from PySide6.QtCore import Qt
from PySide6.QtGui import QAction, QKeySequence
from PySide6.QtWidgets import (
QMainWindow,
QSplitter,
QStatusBar,
QToolBar,
QVBoxLayout,
QWidget
)
from .settings import get_settings
from trovedb import trove as tr
from .note_browser import NoteBrowser
from .note_tool_stack import NoteToolStack
class TroveMainWindow(QMainWindow):
def __init__(self, trove: tr.Trove):
super().__init__()
self.setWindowTitle("Trove")
# ── Toolbar ──
toolbar = QToolBar("Main")
toolbar.setObjectName("maintoolbar")
toolbar.setMovable(False)
self.addToolBar(toolbar)
new_action = QAction("New", self)
new_action.setShortcut(QKeySequence.StandardKey.New)
toolbar.addAction(new_action)
save_action = QAction("Save", self)
save_action.setShortcut(QKeySequence.StandardKey.Save)
toolbar.addAction(save_action)
toolbar.addSeparator()
# ── Central layout ──
central = QWidget()
self.setCentralWidget(central)
layout = QVBoxLayout(central)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
# Horizontal splitter: tree | editor
self._splitter = QSplitter(Qt.Orientation.Horizontal)
self._note_browser = NoteBrowser(trove)
self._splitter.addWidget(self._note_browser)
self._tool_stack = NoteToolStack()
self._splitter.addWidget(self._tool_stack)
layout.addWidget(self._splitter, stretch=1)
self._note_browser.activeNoteChanged.connect(self._tool_stack.onNoteSelected)
# ── Status bar ──
self.setStatusBar(QStatusBar())
self.statusBar().showMessage("Ready")
def closeEvent(self, event):
settings = get_settings()
settings.setValue("window/geometry", self.saveGeometry())
settings.setValue("window/state", self.saveState())
super().closeEvent(event)
def restore_settings(self):
settings = get_settings()
geometry = settings.value("window/geometry")
state = settings.value("window/state")
if geometry:
self.restoreGeometry(geometry)
if state:
self.restoreState(state)

View file

@ -0,0 +1,131 @@
from __future__ import annotations
from typing import Optional, Callable
from PySide6.QtCore import Qt, QModelIndex, Signal
from PySide6.QtWidgets import (
QWidget, QTreeView, QVBoxLayout, QHBoxLayout,
QPushButton, QMenu, QAbstractItemView,
)
from PySide6.QtGui import QAction
from trovedb.trove import Note, TreeNote, Trove
from .trove_tree_model import TroveTreeModel, TroveNode
class NoteBrowser(QWidget):
"""
A simple tree browser widget for a Trove.
Signals:
note_selected(Note) -- emitted when a non-tree note is activated
tree_selected(TreeNote) -- emitted when a tree note is activated
"""
activeNoteChanged = Signal(object) # Note
def __init__(self, trove: Trove, parent: Optional[QWidget] = None):
super().__init__(parent)
self._model = TroveTreeModel(trove)
self._setup_ui()
# ------------------------------------------------------------------
# UI setup
# ------------------------------------------------------------------
def _setup_ui(self) -> None:
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(2)
self._activeNote: Optional[Note] = None
self._tree = QTreeView()
self._tree.setModel(self._model)
self._tree.setHeaderHidden(True)
self._tree.setUniformRowHeights(True)
self._tree.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
self._tree.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
self._tree.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
self._tree.selectionModel().currentChanged.connect(self._on_current_changed)
self._tree.customContextMenuRequested.connect(self._on_context_menu)
layout.addWidget(self._tree)
# Minimal toolbar — refresh only for now
toolbar = QHBoxLayout()
toolbar.setContentsMargins(0, 0, 0, 0)
refresh_btn = QPushButton("Refresh")
refresh_btn.setFixedWidth(70)
refresh_btn.clicked.connect(self._on_refresh)
toolbar.addWidget(refresh_btn)
toolbar.addStretch()
layout.addLayout(toolbar)
# ------------------------------------------------------------------
# Slots
# ------------------------------------------------------------------
def _on_current_changed(self, index: QModelIndex) -> None:
note = index.data(Qt.ItemDataRole.UserRole)
self._handle_set_active_note(note)
def _on_context_menu(self, pos) -> None:
index = self._tree.indexAt(pos)
menu = QMenu(self)
refresh_action = QAction("Refresh", self)
refresh_action.triggered.connect(
lambda: self._model.invalidate_node(index) if index.isValid() else self._on_refresh()
)
menu.addAction(refresh_action)
# Placeholder hooks — wire up to a controller when ready
if index.isValid():
node: TroveNode = index.internalPointer()
if node.is_tree():
menu.addSeparator()
menu.addAction(QAction("New Note…", self)) # TODO: controller
menu.addAction(QAction("New Folder…", self)) # TODO: controller
menu.addSeparator()
menu.addAction(QAction("Rename…", self)) # TODO: controller
menu.addAction(QAction("Delete", self)) # TODO: controller
menu.exec(self._tree.viewport().mapToGlobal(pos))
def _on_refresh(self) -> None:
"""Full model reset — use until event-based invalidation is wired up."""
self._model.invalidate_node(QModelIndex())
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def _handle_set_active_note(self, note: Note | None) -> None:
"""Called by the controller to update the active note."""
if not isinstance(note, Note):
note = None
if self._activeNote != note:
self._activeNote = note
self.activeNoteChanged.emit(note)
def select_note(self, note: Note) -> None:
"""Programmatically select the node matching the given note object."""
match = self._find_index(self._model.index(0, 0, QModelIndex()), note)
if match is not None and match.isValid():
self._tree.setCurrentIndex(match)
self._tree.scrollTo(match)
def _find_index(self, start: QModelIndex, target: Note) -> Optional[QModelIndex]:
"""DFS over visible model nodes looking for a matching Note."""
if not start.isValid():
return None
node: TroveNode = start.internalPointer()
if node.note is target:
return start
for row in range(self._model.rowCount(start)):
child = self._model.index(row, 0, start)
result = self._find_index(child, target)
if result is not None:
return result
return None

View file

@ -0,0 +1,32 @@
from typing import Generator
from PySide6.QtCore import Property, Slot
from PySide6.QtWidgets import QStackedWidget, QWidget, QLabel
import trovedb.trove as tr
from .tool import Tool
from .tool_basic_editor import ToolBasicEditor
class NoteToolStack(QStackedWidget):
def __init__(self, parent: QWidget | None = None):
super().__init__(parent)
self.setContentsMargins(0, 0, 0, 0)
self.addWidget(QLabel("No note selected"))
@Slot(object)
def onNoteSelected(self, note: tr.Note):
for tool in self._iter_tools():
if tool.note == note:
self.setCurrentWidget(tool)
return
tool = ToolBasicEditor(note)
self.addWidget(tool)
self.setCurrentWidget(tool)
def _iter_tools(self) -> Generator[Tool, None, None]:
for i in range(self.count()):
widget = self.widget(i)
if isinstance(widget, Tool):
yield widget

32
trovedb/qgui/qtrove.py Normal file
View file

@ -0,0 +1,32 @@
#!/usr/bin/env python3
"""PySide6 GUI for Trove"""
import sys
from trovedb import trove_factory, user_env
from PySide6.QtWidgets import (
QApplication,
QMessageBox,
)
from .main_window import TroveMainWindow
def main():
app = QApplication(sys.argv)
trove = trove_factory.get_trove(user_env.TROVEBASE) if user_env.TROVEBASE else None
if trove is None:
QMessageBox.critical(None, "Error", "Trove Database Not Found")
sys.exit(1)
# Respect system theme on KDE
app.setStyle("Fusion")
window = TroveMainWindow(trove)
window.show()
window.restore_settings()
sys.exit(app.exec())
if __name__ == '__main__':
main()

5
trovedb/qgui/settings.py Normal file
View file

@ -0,0 +1,5 @@
from PySide6.QtCore import QSettings
def get_settings() -> QSettings:
return QSettings("trovedb", "qtrove")

14
trovedb/qgui/tool.py Normal file
View file

@ -0,0 +1,14 @@
"""Tool Base Class"""
from PySide6.QtWidgets import QWidget
import trovedb.trove as tr
class Tool(QWidget):
def __init__(self, note: tr.Note, parent=None):
super().__init__(parent)
self._note = note
@property
def note(self) -> tr.Note:
return self._note

View file

@ -0,0 +1,28 @@
"""Tool Supporting Basic Editor Functions"""
from typing import cast, Protocol
from PySide6.QtWidgets import QTextEdit, QVBoxLayout
import trovedb.trove as tr
from .tool import Tool
class ToolBasicEditor(Tool):
def __init__(self, note, parent=None):
super().__init__(note, parent)
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
self._text_edit = QTextEdit()
layout.addWidget(self._text_edit)
self.refresh()
def _refresh_blob(self, note: tr.Blob):
self._text_edit.setPlainText(note.read().decode("utf-8"))
def refresh(self):
if isinstance(self.note, tr.Blob):
self._refresh_blob(cast(tr.Blob, self.note))

View file

@ -0,0 +1,129 @@
from typing import Optional
from PySide6.QtCore import QAbstractItemModel, QModelIndex, Qt, QObject
from trovedb.trove import Note, TreeNote, Trove, ObjectId, TreeEntry
class TroveNode:
"""Wrapper around a Note, caching children for tree model use."""
def __init__(self, note: Note, name: str, parent: 'Optional[TroveNode]' = None):
self.note = note
self.name = name
self.parent = parent
self._children: Optional[list[TroveNode]] = None # None = not yet loaded
def is_tree(self) -> bool:
return isinstance(self.note, TreeNote)
def children(self) -> 'list[TroveNode]':
"""Lazy-load and cache children. Sorted by name."""
if self._children is None:
if self.is_tree():
tree: TreeNote = self.note # type: ignore[assignment]
entries: list[TreeEntry] = sorted(tree.entries(), key=lambda e: e.name)
self._children = [
TroveNode(tree.child(e.name), e.name, parent=self)
for e in entries
]
else:
self._children = []
return self._children
def invalidate(self) -> None:
"""Clear cached children, forcing a reload on next access."""
self._children = None
def child_at(self, row: int) -> 'Optional[TroveNode]':
kids = self.children()
return kids[row] if 0 <= row < len(kids) else None
def row(self) -> int:
if self.parent is None:
return 0
return self.parent.children().index(self)
def child_count(self) -> int:
return len(self.children())
class TroveTreeModel(QAbstractItemModel):
"""
Read-only tree model backed by a Trove instance.
The root TreeNote is treated as an invisible root; its children
are the top-level items in the view.
"""
def __init__(self, trove: Trove, parent: Optional[QObject] = None):
super().__init__(parent)
root_note = trove.get_root()
self._root = TroveNode(root_note, "/", parent=None)
# ------------------------------------------------------------------
# QAbstractItemModel interface
# ------------------------------------------------------------------
def index(self, row: int, column: int, parent: QModelIndex = QModelIndex()) -> QModelIndex:
if not self.hasIndex(row, column, parent):
return QModelIndex()
parent_node = self._node_from_index(parent)
child = parent_node.child_at(row)
if child is None:
return QModelIndex()
return self.createIndex(row, column, child)
def parent(self, index: QModelIndex) -> QModelIndex: # type: ignore[override]
if not index.isValid():
return QModelIndex()
node: TroveNode = index.internalPointer() # type: ignore[assignment]
parent_node = node.parent
if parent_node is None or parent_node is self._root:
return QModelIndex()
return self.createIndex(parent_node.row(), 0, parent_node)
def rowCount(self, parent: QModelIndex = QModelIndex()) -> int:
parent_node = self._node_from_index(parent)
return parent_node.child_count()
def columnCount(self, parent: QModelIndex = QModelIndex()) -> int:
return 1
def data(self, index: QModelIndex, role: int = Qt.DisplayRole) -> object: # type: ignore[assignment]
if not index.isValid():
return None
node: TroveNode = index.internalPointer() # type: ignore[assignment]
if role == Qt.DisplayRole:
return node.name
if role == Qt.UserRole:
return node.note
return None
def hasChildren(self, parent: QModelIndex = QModelIndex()) -> bool:
node = self._node_from_index(parent)
return node.is_tree()
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _node_from_index(self, index: QModelIndex) -> TroveNode:
if index.isValid():
node = index.internalPointer()
if node is not None:
return node # type: ignore[return-value]
return self._root
def invalidate_node(self, index: QModelIndex) -> None:
"""
Force a reload of the node at index.
Call this after mutations to the underlying Trove tree.
"""
node = self._node_from_index(index)
self.beginResetModel()
node.invalidate()
self.endResetModel()

View file

@ -9,6 +9,9 @@ only the data field — storage is the caller's concern.
import json
from .trove import ObjectId
class Tree:
def __init__(self, data: bytes | None = None):
"""
@ -16,7 +19,7 @@ class Tree:
An empty Tree is created if data is None.
"""
if not data:
self._entries: dict[str, int] = {}
self._entries: dict[str, ObjectId] = {}
else:
self._entries = json.loads(data.decode("utf-8"))
@ -24,11 +27,11 @@ class Tree:
"""Serialize the tree to UTF-8 JSON bytes."""
return json.dumps(self._entries).encode("utf-8")
def set_entry(self, name: str, object_id: int) -> None:
def set_entry(self, name: str, object_id: ObjectId) -> None:
"""Add or update an entry mapping name -> uuid."""
self._entries[name] = object_id
def get_entry(self, name: str) -> int:
def get_entry(self, name: str) -> ObjectId:
"""Get the uuid associated with a name, or raise KeyError if not found."""
return self._entries[name]
@ -36,6 +39,6 @@ class Tree:
"""Remove an entry by name. Raises KeyError if not found."""
del self._entries[name]
def list(self) -> dict[str, int]:
def list(self) -> dict[str, ObjectId]:
"""Return a shallow copy of all entries as {name: uuid}."""
return dict(self._entries)

View file

@ -1,6 +1,7 @@
from typing import Protocol, runtime_checkable, Optional, Dict, List, Self, NamedTuple, Iterable, TypedDict
from uuid import UUID
from pathlib import PurePosixPath
import datetime as dt
type ObjectId = int | str | UUID
@ -26,11 +27,27 @@ class Note(Protocol):
Protocol for a Note item.
Represents access to an individual note's content and metadata.
"""
@property
def object_id(self) -> ObjectId:
"""The unique identifier for this note."""
...
@property
def mime(self) -> str:
"""The MIME type of the note's content."""
...
@property
def readonly(self) -> bool:
"""Whether the note is read-only."""
...
@property
def mtime(self) -> dt.datetime:
"""The last modification time of the note."""
...
def get_raw_metadata(self, key: str) -> Optional[bytes]:
"""Retrieve metadata value for the given key."""
...
@ -84,10 +101,12 @@ class Tree(Protocol):
"""Return all entries as {name: object_id}."""
...
class BlobNote(Note, Blob):
@runtime_checkable
class BlobNote(Note, Blob, Protocol):
"""Blob Note"""
class TreeNote(Note, Tree):
@runtime_checkable
class TreeNote(Note, Tree, Protocol):
"""Tree Note"""

View file

@ -8,28 +8,43 @@ 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
from .trove import Note, Trove, TreeNote, BlobNote, TreeEntry, NoteNotFound, ObjectId
class NoteImpl(Note):
"""Concrete not implementation"""
def __init__(self, parent: 'TroveImpl', object_id: int):
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) -> int:
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 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)
@ -108,7 +123,7 @@ class TreeNoteImpl(NoteImpl, TreeNote):
for name, object_id in tree.list().items():
yield TreeEntry(name, object_id)
def list(self) -> dict[str, int]:
def list(self) -> dict[str, ObjectId]:
"""Return all entries as {name: object_id}."""
return self._read_tree().list()
@ -153,7 +168,7 @@ class TroveImpl:
self.close()
# Trove protocol
def get_raw_note(self, note_id: int) -> Note:
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: