andrew/trove
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: andrew:82c272990ca1b9a9146f059e7a5bd5ac1787bb87
Branches Tags
andrew:main
..
compare: andrew:22a9c686114b84625c7339080dcb5b6dad6e5309
Branches Tags
andrew:main

No commits in common. "82c272990ca1b9a9146f059e7a5bd5ac1787bb87" and "22a9c686114b84625c7339080dcb5b6dad6e5309" have entirely different histories.
82c272990c ... 22a9c68611

15 changed files with 59 additions and 585 deletions
Download patch file Download diff file Expand all files Collapse all files

1
pyproject.toml
View file

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

94
trovedb/db.py
View file

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

24
trovedb/fs.py
View file

@ -1,7 +1,6 @@
import os import os
import sqlite3 import sqlite3
import tempfile import tempfile
import datetime as dt
from pathlib import Path from pathlib import Path
from typing import Optional, Dict, List, Self, Iterable from typing import Optional, Dict, List, Self, Iterable
from .trove import Note, Trove, TreeNote, BlobNote, Blob, Tree, BadNoteType, TreeEntry, NoteNotFound 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") raise ValueError("Note not yet saved to disk")
return self._inode 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 @property
def _path(self) -> Path: def _path(self) -> Path:
if self._fs_path is not None: if self._fs_path is not None:
@ -79,11 +60,6 @@ class FSBlobNote(FSNote, BlobNote):
pass pass
class FSTreeNote(FSNote, TreeNote): 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): def link(self, name: str, note: Note):
if not isinstance(note, FSBlobNote): if not isinstance(note, FSBlobNote):
raise BadNoteType("Only blob notes can be linked") raise BadNoteType("Only blob notes can be linked")

14
trovedb/fuse/server.py
View file

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

77
trovedb/qgui/main_window.py
View file

@ -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)

131
trovedb/qgui/note_browser.py
View file

@ -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

32
trovedb/qgui/note_tool_stack.py
View file

@ -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

32
trovedb/qgui/qtrove.py
View file

@ -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()

5
trovedb/qgui/settings.py
View file

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

14
trovedb/qgui/tool.py
View file

@ -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

28
trovedb/qgui/tool_basic_editor.py
View file

@ -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))

129
trovedb/qgui/trove_tree_model.py
View file

@ -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()

11
trovedb/tree.py
View file

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

23
trovedb/trove.py
View file

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

25
trovedb/trovedb.py
View file

@ -8,43 +8,28 @@ tree serialization respectively.
from typing import Optional, Self from typing import Optional, Self
from pathlib import Path from pathlib import Path
import datetime as dt
from .db import Sqlite3Trove, NOTE_ROOT_ID from .db import Sqlite3Trove, NOTE_ROOT_ID
from .tree import Tree as TreeData from .tree import Tree as TreeData
from . import trove as tr 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): class NoteImpl(Note):
"""Concrete not implementation""" """Concrete not implementation"""
def __init__(self, parent: 'TroveImpl', object_id: ObjectId): def __init__(self, parent: 'TroveImpl', object_id: int):
self._parent = parent self._parent = parent
self._db = parent.db self._db = parent.db
self._object_id = object_id self._object_id = object_id
# Note protocol # Note protocol
@property @property
def object_id(self) -> ObjectId: def object_id(self) -> int:
return self._object_id 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]: def get_raw_metadata(self, key: str) -> Optional[bytes]:
return self._db.read_metadata(self._object_id, key) 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(): for name, object_id in tree.list().items():
yield TreeEntry(name, object_id) 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 all entries as {name: object_id}."""
return self._read_tree().list() return self._read_tree().list()
@ -168,7 +153,7 @@ class TroveImpl:
self.close() self.close()
# Trove protocol # 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.""" """Return a BlobNote or TreeNote for the given id, or None if not found."""
ot = self._db.get_object_type(note_id) ot = self._db.get_object_type(note_id)
if ot is None: if ot is None: