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

Switch to UUID for sqlite db

Browse source
This commit is contained in:
Andrew Mulbrook 2026-03-24 21:22:24 -05:00
parent 90b72ed678
commit 82c272990c
3 changed files with 59 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

93
trovedb/db.py
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"", NOTE_ROOT_ID)
obj.write_tree(b"", NOTE_ROOT_ID)
return obj
def close(self):
@ -87,95 +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 get_mtime(self, object_id: int) -> datetime | 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 = ?", (object_id,)
"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: int, key: str) -> bytes | None:
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.
"""
@ -184,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()
@ -207,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.
"""
@ -229,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")