diff --git a/trovedb/db.py b/trovedb/db.py new file mode 100644 index 0000000..62e5508 --- /dev/null +++ b/trovedb/db.py @@ -0,0 +1,296 @@ +""" +Python API for accessing underlying SQlite3 File Database + +A Sqlite3 database provides the basic underlying interface to the underlying +notes interface. The low level database is an extremely simple model using +similar ideas to get storage. However, UUID are utilized for persistent object +references. +""" + +import argparse +import sqlite3 +import sys +import uuid as _uuid +from datetime import datetime, timezone +from pathlib import Path + + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS objects ( + uuid TEXT PRIMARY KEY, + type TEXT NOT NULL CHECK(type IN ('blob', 'tree')), + data BLOB, + created TEXT NOT NULL, + modified TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS metadata ( + uuid TEXT NOT NULL REFERENCES objects(uuid) ON DELETE CASCADE, + key TEXT NOT NULL, + value BLOB, + PRIMARY KEY (uuid, key) +); + +CREATE TABLE IF NOT EXISTS labels +( + label + TEXT + PRIMARY + KEY, + uuid + TEXT + NOT + NULL + REFERENCES + objects +( + uuid +) ON DELETE CASCADE + ); \ + """ + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +class TroveDB: + def __init__(self, con: sqlite3.Connection): + self._con = con + self._con.execute("PRAGMA foreign_keys = ON") + self._con.row_factory = sqlite3.Row + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + @classmethod + def init(cls, path: str | Path) -> "TroveDB": + """Create a new Trove database at path. Fails if file already exists.""" + p = Path(path) + if p.exists(): + raise FileExistsError(f"Database already exists: {p}") + con = sqlite3.connect(str(p)) + con.executescript(_SCHEMA) + con.commit() + return cls(con) + + @classmethod + def open(cls, path: str | Path) -> "TroveDB": + """Open an existing Trove database.""" + p = Path(path) + if not p.exists(): + raise FileNotFoundError(f"Database not found: {p}") + con = sqlite3.connect(str(p)) + return cls(con) + + def close(self): + self._con.close() + + def __enter__(self): + return self + + def __exit__(self, *_): + self.close() + + # ------------------------------------------------------------------ + # CRUD operations + # ------------------------------------------------------------------ + + def read_blob(self, uuid: str) -> bytes | None: + """Return raw data for a blob object, or None if not found.""" + row = self._con.execute( + "SELECT data, type FROM objects WHERE uuid = ?", (uuid,) + ).fetchone() + if row is None: + return None + if row["type"] != "blob": + raise TypeError(f"Object {uuid} is type '{row['type']}', not 'blob'") + return bytes(row["data"]) if row["data"] is not None else b"" + + def read_metadata(self, uuid: str, 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 uuid = ? AND key = ?", (uuid, key) + ).fetchone() + if row is None: + return None + return bytes(row["value"]) if row["value"] is not None else b"" + + def write_blob(self, data: bytes, existing_uuid: str | None = None) -> str: + """ + Insert or replace a blob. Returns the uuid. + Pass existing_uuid to update an existing object. + """ + now = _now() + uid = existing_uuid or str(_uuid.uuid4()) + if existing_uuid: + self._con.execute( + "UPDATE objects SET data = ?, modified = ? WHERE uuid = ? AND type = 'blob'", + (data, now, uid), + ) + else: + self._con.execute( + "INSERT INTO objects (uuid, type, data, created, modified) VALUES (?, 'blob', ?, ?, ?)", + (uid, data, now, now), + ) + self._con.commit() + return uid + + def delete_blob(self, uuid: str) -> bool: + """ + Delete a blob and all its metadata rows. + Returns True if an object was deleted, False if uuid not found. + Foreign key cascade handles the metadata rows. + """ + cur = self._con.execute( + "DELETE FROM objects WHERE uuid = ?", (uuid,) + ) + self._con.commit() + return cur.rowcount > 0 + + def get_label(self, label: str) -> str | None: + """ + Return the UUID associated with a label, or None if not found. + """ + row = self._con.execute( + "SELECT uuid FROM labels WHERE label = ?", (label,) + ).fetchone() + if row is None: + return None + return row["uuid"] + + def set_label(self, label: str, uuid: str) -> None: + """ + Set a label to point to a UUID. Creates or updates the label. + The UUID must exist in the objects table. + """ + self._con.execute( + "INSERT OR REPLACE INTO labels (label, uuid) VALUES (?, ?)", + (label, uuid), + ) + self._con.commit() + + def delete_label(self, label: str) -> bool: + """ + Delete a label. Returns True if a label was deleted, False if not found. + """ + cur = self._con.execute( + "DELETE FROM labels WHERE label = ?", (label,) + ) + self._con.commit() + return cur.rowcount > 0 + + def list_labels(self) -> list[tuple[str, str]]: + """ + Return all labels as a list of (label, uuid) tuples. + """ + rows = self._con.execute( + "SELECT label, uuid FROM labels ORDER BY label" + ).fetchall() + return [(row["label"], row["uuid"]) for row in rows] + + +def main(): + """Command-line interface for TroveDB.""" + parser = argparse.ArgumentParser(description="TroveDB command-line interface") + parser.add_argument("database", help="Path to the database file") + + subparsers = parser.add_subparsers(dest="operation", required=True, help="Operation to perform") + + # Create database + subparsers.add_parser("create", help="Create a new database") + + # Get blob + get_parser = subparsers.add_parser("get", help="Get a blob object by UUID") + get_parser.add_argument("uuid", help="UUID 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("--uuid", help="UUID of existing blob to update (optional)") + + # Delete blob + delete_parser = subparsers.add_parser("delete", help="Delete a blob by UUID") + delete_parser.add_argument("uuid", help="UUID of the blob to delete") + + # Set label + setlabel_parser = subparsers.add_parser("setlabel", help="Create or update a label to point to a UUID") + setlabel_parser.add_argument("label", help="Label name") + setlabel_parser.add_argument("uuid", help="UUID to associate with the label") + + # Remove label + rmlabel_parser = subparsers.add_parser("rmlabel", help="Delete a label") + rmlabel_parser.add_argument("label", help="Label name to delete") + + # List labels + subparsers.add_parser("labels", help="List all labels") + + args = parser.parse_args() + + try: + match args.operation: + case "create": + db = TroveDB.init(args.database) + db.close() + print(f"Database created: {args.database}") + + case "get": + db = TroveDB.open(args.database) + data = db.read_blob(args.uuid) + db.close() + if data is None: + print(f"Blob not found: {args.uuid}", file=sys.stderr) + sys.exit(1) + sys.stdout.buffer.write(data) + + case "write": + data_bytes = args.data.encode("utf-8") + db = TroveDB.open(args.database) + uuid = db.write_blob(data_bytes, args.uuid) + db.close() + print(uuid) + + case "delete": + db = TroveDB.open(args.database) + deleted = db.delete_blob(args.uuid) + db.close() + if deleted: + print(f"Deleted blob: {args.uuid}") + else: + print(f"Blob not found: {args.uuid}", file=sys.stderr) + sys.exit(1) + + case "setlabel": + db = TroveDB.open(args.database) + db.set_label(args.label, args.uuid) + db.close() + print(f"Label '{args.label}' set to UUID: {args.uuid}") + + case "rmlabel": + db = TroveDB.open(args.database) + deleted = db.delete_label(args.label) + db.close() + if deleted: + print(f"Deleted label: {args.label}") + else: + print(f"Label not found: {args.label}", file=sys.stderr) + sys.exit(1) + + case "labels": + db = TroveDB.open(args.database) + labels = db.list_labels() + db.close() + if labels: + for label, uuid in labels: + print(f"{label}: {uuid}") + else: + print("No labels found.") + + except Exception as e: + print(f"Error: {e}", file=sys.stderr) + sys.exit(1) + + +if __name__ == "__main__": + main()