db: Add objects for Tables and Rows

Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>

db/objects.py

@@ -0,0 +1,63 @@
+# Copyright 2021 (c) Anna Schumaker.
+import sqlite3
+from gi.repository import GObject
+from . import execute
+
+class Row(GObject.GObject):
+    def __init__(self, rowid):
+        GObject.GObject.__init__(self)
+        self.rowid = rowid
+
+    def __eq__(self, rhs): return self.rowid == rhs.rowid
+    def __int__(self): return self.rowid
+
+    def do_get_column(self, column): raise NotImplementedError
+
+
+class Table(GObject.GObject):
+    def __init__(self, name, type):
+        GObject.GObject.__init__(self)
+        self.table_name = name
+        self.table_type = type
+
+        self.do_create()
+
+    def __getitem__(self, rowid): return self.get(rowid)
+
+    def do_create(self): raise NotImplementedError
+    def do_delete(self, obj): raise NotImplementedError
+    def do_insert(self, *args): raise NotImplementedError
+    def do_get(self, rowid): raise NotImplementedError
+    def do_lookup(self, *args): raise NotImplementedError
+
+    def insert(self, *args):
+        cursor = self.do_insert(*args)
+        object = self.table_type(cursor.lastrowid)
+        self.emit("row-inserted", object)
+        return object
+
+    def delete(self, obj):
+        self.emit("row-deleted", obj)
+        self.do_delete(obj)
+
+    def find(self, *args):
+        if object := self.lookup(*args):
+            return object
+        return self.insert(*args)
+
+    def get(self, rowid):
+        row = self.do_get(rowid).fetchone()
+        return self.table_type(row[0]) if row else None
+
+    def lookup(self, *args):
+        row = self.do_lookup(*args).fetchone()
+        return self.table_type(row[0]) if row else None
+
+    def drop(self):
+        execute(f"DROP TABLE {self.table_name}")
+
+    @GObject.Signal(arg_types=(Row,))
+    def row_inserted(self, row): pass
+
+    @GObject.Signal(arg_types=(Row,))
+    def row_deleted(self, row): pass

db/test_objects.py

@@ -0,0 +1,76 @@
+# Copyright 2021 (c) Anna Schumaker.
+import unittest
+from gi.repository import GObject
+from . import objects
+
+class FakeCursor:
+    def __init__(self, lastrowid): self.lastrowid = lastrowid
+    def fetchone(self): return (self.lastrowid,) if self.lastrowid else None
+
+
+class TestRow(unittest.TestCase):
+    def test_row(self):
+        row = objects.Row(1)
+        self.assertEqual(row.rowid, 1)
+        self.assertIsInstance(row, GObject.GObject)
+
+        self.assertEqual(int(row), 1)
+        self.assertEqual(row, objects.Row(1))
+        self.assertNotEqual(row, objects.Row(2))
+
+        with self.assertRaises(NotImplementedError):
+            row.do_get_column("test")
+
+
+class FakeTable(objects.Table):
+    def do_create(self): self.create_called = True
+    def do_insert(self, insert, args): return FakeCursor(1)
+    def do_delete(self, obj): self.delete_called = True
+    def do_get(self, rowid): return FakeCursor(1)
+    def do_lookup(self, lookup, args):
+        return FakeCursor(1) if lookup == "lookup" else FakeCursor(None)
+
+class TestTable(unittest.TestCase):
+    def on_row_inserted(self, table, row):
+        self.row_inserted = row
+
+    def on_row_deleted(self, table, row):
+        self.row_deleted = row
+
+    def test_table_errors(self):
+        with self.assertRaises(NotImplementedError):
+            table = objects.Table("table", objects.Row)
+        with self.assertRaises(NotImplementedError):
+            objects.Table.do_delete(None, objects.Row(1))
+        with self.assertRaises(NotImplementedError):
+            objects.Table.do_insert(None, "insert", "arguments")
+        with self.assertRaises(NotImplementedError):
+            objects.Table.do_get(None, 2)
+        with self.assertRaises(NotImplementedError):
+            objects.Table.do_lookup(None, "lookup", "arguments")
+
+    def test_table(self):
+        table = FakeTable("fake", objects.Row)
+        table.connect("row-inserted", self.on_row_inserted)
+        table.connect("row-deleted", self.on_row_deleted)
+
+        self.assertIsInstance(table, GObject.GObject)
+        self.assertEqual(table.table_name, "fake")
+        self.assertEqual(table.table_type, objects.Row)
+        self.assertTrue(table.create_called)
+
+        self.assertEqual(table.insert("insert", "arguments"), objects.Row(1))
+        self.assertEqual(self.row_inserted, objects.Row(1))
+
+        table.delete(objects.Row(1))
+        self.assertEqual(self.row_deleted, objects.Row(1))
+        self.assertTrue(table.delete_called)
+
+        self.assertEqual(table.get(1), objects.Row(1))
+        self.assertEqual(table[1], objects.Row(1))
+
+        self.assertEqual(table.lookup("lookup", "arguments"), objects.Row(1))
+
+        self.row_inserted = None
+        self.assertEqual(table.find("find", "arguments"), objects.Row(1))
+        self.assertEqual(self.row_inserted, objects.Row(1))