diff --git a/pyproject.toml b/pyproject.toml index 90b7c61..39c7876 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,6 @@ dependencies = [] [project.scripts] trove = "trovedb.cli:cli_main" -qtrove = "trovedb.qgui.qtrove:main" [project.optional-dependencies] dev = [ diff --git a/trovedb/db.py b/trovedb/db.py index a4d846f..e811028 100644 --- a/trovedb/db.py +++ b/trovedb/db.py @@ -9,22 +9,21 @@ 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 = uuid.UUID(int=0) +NOTE_ROOT_ID = 1 _SCHEMA = """ CREATE TABLE IF NOT EXISTS objects ( - id TEXT PRIMARY KEY, + id INTEGER PRIMARY KEY, type TEXT NOT NULL CHECK(type IN ('blob', 'tree')), data BLOB, modified TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS metadata ( - id TEXT NOT NULL REFERENCES objects(id) ON DELETE CASCADE, + id INTEGER NOT NULL REFERENCES objects(id) ON DELETE CASCADE, key TEXT NOT NULL, value BLOB, PRIMARY KEY (id, key) @@ -33,21 +32,14 @@ CREATE TABLE IF NOT EXISTS metadata ( CREATE TABLE IF NOT EXISTS labels ( label TEXT PRIMARY KEY, - id TEXT NOT NULL REFERENCES objects(id) ON DELETE CASCADE + id INTEGER 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) @@ -79,7 +71,7 @@ class Sqlite3Trove: con.commit() obj = cls(con) if initialize: - obj.write_tree(b"", NOTE_ROOT_ID) + obj.write_blob(b"", NODE_ROOT_ID) return obj def close(self): @@ -95,88 +87,86 @@ class Sqlite3Trove: # CRUD operations # ------------------------------------------------------------------ - def get_object_type(self, object_id: SqlObjectId) -> str | None: + def get_object_type(self, object_id: int) -> str | None: """Return the type column for an object, or None if not found.""" row = self._con.execute( - "SELECT type FROM objects WHERE id = ?", (_sql_id(object_id),) + "SELECT type FROM objects WHERE id = ?", (object_id,) ).fetchone() return row["type"] if row else None - def read_object(self, object_id: SqlObjectId) -> bytes | None: + def read_object(self, object_id: int) -> 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 = ?", (_sql_id(object_id),) + "SELECT data, type FROM objects WHERE id = ?", (object_id,) ).fetchone() if row is None: return None return bytes(row["data"]) if row["data"] is not None else b"" - 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: + def read_metadata(self, object_id: int, 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 = ?", (_sql_id(object_id), key) + "SELECT value FROM metadata WHERE id = ? AND key = ?", (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: SqlObjectId, key: str, value: bytes) -> None: + def write_metadata(self, object_id: int, 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 (?, ?, ?)", - (_sql_id(object_id), key, value), + (object_id, key, value), ) self._con.commit() - def _write_object(self, data: bytes, dtype: str, object_id: str | uuid.UUID | None = None) -> str: + def _write_object(self, data: bytes, dtype: str, object_id: int | None = None) -> int: """ Insert or replace an object. Returns the id. - If object_id is None, creates a new object with a new UUID. + If object_id is None, creates a new object with auto-assigned ID. If object_id is provided, updates or creates the object with that ID. """ modified = _now() if object_id is None: - 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) + 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 - def write_blob(self, data: bytes, object_id: SqlObjectId | None = None) -> str: + def write_blob(self, data: bytes, object_id: int | None = None) -> int: """ Insert or replace a blob. Returns the id. Pass object_id to update an existing object. """ - return self._write_object(data, "blob", _sql_id(object_id)) + return self._write_object(data, "blob", object_id) - def write_tree(self, data: bytes, object_id: SqlObjectId | None = None) -> str: + def write_tree(self, data: bytes, object_id: int | None = None) -> int: """Write a tree-typed object. Returns the assigned id.""" - return self._write_object(data, "tree", _sql_id(object_id)) + return self._write_object(data, "tree", object_id) - def delete_object(self, object_id: SqlObjectId) -> bool: + def delete_object(self, object_id: int) -> 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 = ?", (_sql_id(object_id),) + "DELETE FROM objects WHERE id = ?", (object_id,) ) self._con.commit() return cur.rowcount > 0 - def get_label(self, label: str) -> SqlObjectId | None: + def get_label(self, label: str) -> int | None: """ Return the ID associated with a label, or None if not found. """ @@ -185,16 +175,16 @@ class Sqlite3Trove: ).fetchone() if row is None: return None - return uuid.UUID(row["id"]) + return row["id"] - def set_label(self, label: str, object_id: SqlObjectId) -> None: + def set_label(self, label: str, object_id: int) -> 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, _sql_id(object_id)), + (label, object_id), ) self._con.commit() @@ -208,7 +198,7 @@ class Sqlite3Trove: self._con.commit() return cur.rowcount > 0 - def list_labels(self) -> list[tuple[str, str]]: + def list_labels(self) -> list[tuple[str, int]]: """ Return all labels as a list of (label, id) tuples. """ @@ -230,21 +220,21 @@ def main(): # Get blob get_parser = subparsers.add_parser("get", help="Get a blob object by ID") - get_parser.add_argument("id", help="ID of the blob to retrieve") + get_parser.add_argument("id", type=int, 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", help="ID of existing blob to update (optional)") + write_parser.add_argument("--id", type=int, 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", help="ID of the blob to delete") + delete_parser.add_argument("id", type=int, 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", help="ID to associate with the label") + setlabel_parser.add_argument("id", type=int, help="ID to associate with the label") # Remove label rmlabel_parser = subparsers.add_parser("rmlabel", help="Delete a label") diff --git a/trovedb/fs.py b/trovedb/fs.py index 3f36792..ff6eece 100644 --- a/trovedb/fs.py +++ b/trovedb/fs.py @@ -1,7 +1,6 @@ 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 @@ -25,24 +24,6 @@ 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: @@ -79,11 +60,6 @@ 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") diff --git a/trovedb/fuse/server.py b/trovedb/fuse/server.py index f933972..2441314 100644 --- a/trovedb/fuse/server.py +++ b/trovedb/fuse/server.py @@ -194,26 +194,20 @@ 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 = mtime_ns - attr.st_ctime_ns = mtime_ns + attr.st_mtime_ns = now_ns + attr.st_ctime_ns = now_ns attr.generation = 0 attr.entry_timeout = 5.0 attr.attr_timeout = 5.0 - - # Determine permissions based on readonly property if is_tree: - mode = 0o755 if not note.readonly else 0o555 - attr.st_mode = stat.S_IFDIR | mode + attr.st_mode = stat.S_IFDIR | 0o755 attr.st_size = 0 attr.st_blksize = 512 attr.st_blocks = 0 else: - mode = 0o644 if not note.readonly else 0o444 - attr.st_mode = stat.S_IFREG | mode + attr.st_mode = stat.S_IFREG | 0o644 attr.st_size = size attr.st_blksize = 512 attr.st_blocks = (size + 511) // 512 diff --git a/trovedb/qgui/main_window.py b/trovedb/qgui/main_window.py deleted file mode 100644 index d4cdde4..0000000 --- a/trovedb/qgui/main_window.py +++ /dev/null @@ -1,77 +0,0 @@ -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) diff --git a/trovedb/qgui/note_browser.py b/trovedb/qgui/note_browser.py deleted file mode 100644 index c672c00..0000000 --- a/trovedb/qgui/note_browser.py +++ /dev/null @@ -1,131 +0,0 @@ -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 \ No newline at end of file diff --git a/trovedb/qgui/note_tool_stack.py b/trovedb/qgui/note_tool_stack.py deleted file mode 100644 index d5cb2c5..0000000 --- a/trovedb/qgui/note_tool_stack.py +++ /dev/null @@ -1,32 +0,0 @@ -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 diff --git a/trovedb/qgui/qtrove.py b/trovedb/qgui/qtrove.py deleted file mode 100644 index e41b1e8..0000000 --- a/trovedb/qgui/qtrove.py +++ /dev/null @@ -1,32 +0,0 @@ -#!/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() \ No newline at end of file diff --git a/trovedb/qgui/settings.py b/trovedb/qgui/settings.py deleted file mode 100644 index 40f9681..0000000 --- a/trovedb/qgui/settings.py +++ /dev/null @@ -1,5 +0,0 @@ -from PySide6.QtCore import QSettings - - -def get_settings() -> QSettings: - return QSettings("trovedb", "qtrove") diff --git a/trovedb/qgui/tool.py b/trovedb/qgui/tool.py deleted file mode 100644 index 53f2829..0000000 --- a/trovedb/qgui/tool.py +++ /dev/null @@ -1,14 +0,0 @@ -"""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 diff --git a/trovedb/qgui/tool_basic_editor.py b/trovedb/qgui/tool_basic_editor.py deleted file mode 100644 index 530b88c..0000000 --- a/trovedb/qgui/tool_basic_editor.py +++ /dev/null @@ -1,28 +0,0 @@ -"""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)) - - diff --git a/trovedb/qgui/trove_tree_model.py b/trovedb/qgui/trove_tree_model.py deleted file mode 100644 index c74612f..0000000 --- a/trovedb/qgui/trove_tree_model.py +++ /dev/null @@ -1,129 +0,0 @@ -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() \ No newline at end of file diff --git a/trovedb/tree.py b/trovedb/tree.py index 3336609..75c64ac 100644 --- a/trovedb/tree.py +++ b/trovedb/tree.py @@ -9,9 +9,6 @@ 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): """ @@ -19,7 +16,7 @@ class Tree: An empty Tree is created if data is None. """ if not data: - self._entries: dict[str, ObjectId] = {} + self._entries: dict[str, int] = {} else: self._entries = json.loads(data.decode("utf-8")) @@ -27,11 +24,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: ObjectId) -> None: + def set_entry(self, name: str, object_id: int) -> None: """Add or update an entry mapping name -> uuid.""" self._entries[name] = object_id - def get_entry(self, name: str) -> ObjectId: + def get_entry(self, name: str) -> int: """Get the uuid associated with a name, or raise KeyError if not found.""" return self._entries[name] @@ -39,6 +36,6 @@ class Tree: """Remove an entry by name. Raises KeyError if not found.""" del self._entries[name] - def list(self) -> dict[str, ObjectId]: + def list(self) -> dict[str, int]: """Return a shallow copy of all entries as {name: uuid}.""" return dict(self._entries) diff --git a/trovedb/trove.py b/trovedb/trove.py index 924331f..c9d0be9 100644 --- a/trovedb/trove.py +++ b/trovedb/trove.py @@ -1,7 +1,6 @@ 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 @@ -27,27 +26,11 @@ 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.""" ... @@ -101,12 +84,10 @@ class Tree(Protocol): """Return all entries as {name: object_id}.""" ... -@runtime_checkable -class BlobNote(Note, Blob, Protocol): +class BlobNote(Note, Blob): """Blob Note""" -@runtime_checkable -class TreeNote(Note, Tree, Protocol): +class TreeNote(Note, Tree): """Tree Note""" diff --git a/trovedb/trovedb.py b/trovedb/trovedb.py index 5e0e877..1cf2f2e 100644 --- a/trovedb/trovedb.py +++ b/trovedb/trovedb.py @@ -8,43 +8,28 @@ 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, ObjectId +from .trove import Note, Trove, TreeNote, BlobNote, TreeEntry, NoteNotFound class NoteImpl(Note): """Concrete not implementation""" - def __init__(self, parent: 'TroveImpl', object_id: ObjectId): + 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) -> ObjectId: + 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) @@ -123,7 +108,7 @@ class TreeNoteImpl(NoteImpl, TreeNote): for name, object_id in tree.list().items(): yield TreeEntry(name, object_id) - def list(self) -> dict[str, ObjectId]: + def list(self) -> dict[str, int]: """Return all entries as {name: object_id}.""" return self._read_tree().list() @@ -168,7 +153,7 @@ class TroveImpl: self.close() # Trove protocol - def get_raw_note(self, note_id: ObjectId) -> Note: + 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: