db: Remove the now unused objects.py file
Implements: Issue #18 (Remove the db.objects.Map class) Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>
This commit is contained in:
parent
932663f872
commit
f6c72ed081
|
|
@ -15,7 +15,6 @@
|
|||
# | genreid | trackid |
|
||||
# +---------+---------+
|
||||
from gi.repository import GObject
|
||||
from . import objects
|
||||
from . import playlist
|
||||
from . import track
|
||||
from . import sql
|
||||
|
|
|
|||
125
db/objects.py
125
db/objects.py
|
|
@ -1,125 +0,0 @@
|
|||
# Copyright 2021 (c) Anna Schumaker.
|
||||
import sqlite3
|
||||
from gi.repository import GObject
|
||||
from .sql 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
|
||||
|
||||
def get_column(self, column):
|
||||
return self.do_get_column(column).fetchone()[column]
|
||||
|
||||
|
||||
class Tag(Row):
|
||||
def __gt__(self, rhs): return self.sort > rhs.sort
|
||||
def __lt__(self, rhs): return self.sort < rhs.sort
|
||||
def __str__(self): return self.name
|
||||
|
||||
@GObject.Property
|
||||
def name(self): return self.get_column("name")
|
||||
|
||||
@GObject.Property
|
||||
def sort(self): return self.get_column("sort")
|
||||
|
||||
|
||||
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_get_all(self): 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 get_all(self):
|
||||
rows = self.do_get_all().fetchall()
|
||||
return [ self.table_type(row[0]) for row in rows ]
|
||||
|
||||
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
|
||||
|
||||
|
||||
class Map(GObject.GObject):
|
||||
def __init__(self, name, lhs, rhs):
|
||||
GObject.GObject.__init__(self)
|
||||
self.map_name = name
|
||||
self.map_lhs = lhs
|
||||
self.map_rhs = rhs
|
||||
|
||||
self.do_create()
|
||||
|
||||
def do_create(self): raise NotImplementedError
|
||||
def do_insert(self, lhs, rhs): raise NotImplementedError
|
||||
def do_delete(self, lhs, rhs): raise NotImplementedError
|
||||
def do_lookup_rhs(self, lhs): raise NotImplementedError
|
||||
def do_lookup_lhs(self, rhs): raise NotImplementedError
|
||||
|
||||
def insert(self, lhs, rhs):
|
||||
cursor = self.do_insert(lhs, rhs)
|
||||
self.emit("row-inserted", lhs, rhs)
|
||||
|
||||
def delete(self, lhs, rhs):
|
||||
self.emit("row-deleted", lhs, rhs)
|
||||
self.do_delete(lhs, rhs)
|
||||
|
||||
def lookup_rhs(self, lhs):
|
||||
cursor = self.do_lookup_rhs(lhs)
|
||||
return [ self.map_rhs(row[0]) for row in cursor.fetchall() ]
|
||||
|
||||
def lookup_lhs(self, rhs):
|
||||
cursor = self.do_lookup_lhs(rhs)
|
||||
return [ self.map_lhs(row[0]) for row in cursor.fetchall() ]
|
||||
|
||||
def reset(self):
|
||||
execute(f"DROP TABLE {self.map_name}")
|
||||
self.do_create()
|
||||
|
||||
@GObject.Signal(arg_types=(GObject.GObject,GObject.GObject))
|
||||
def row_inserted(self, lhs, rhs): pass
|
||||
|
||||
@GObject.Signal(arg_types=(GObject.GObject,GObject.GObject))
|
||||
def row_deleted(self, lhs, rhs): pass
|
||||
|
|
@ -5,7 +5,6 @@
|
|||
# | plstateid | random | loop | current | sort |
|
||||
# +-----------+--------+------+---------+------+
|
||||
from gi.repository import GObject
|
||||
from . import objects
|
||||
from . import sql
|
||||
from . import table
|
||||
|
||||
|
|
|
|||
|
|
@ -1,169 +0,0 @@
|
|||
# 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
|
||||
def fetchall(self): return [ (2,), (3,) ] 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 FakeTagCursor:
|
||||
def __init__(self, col, text):
|
||||
self.col = col
|
||||
self.text = text
|
||||
def fetchone(self): return { self.col : self.text }
|
||||
|
||||
class FakeTag(objects.Tag):
|
||||
def __init__(self, tid, name, sort):
|
||||
objects.Tag.__init__(self, tid)
|
||||
self.fake_name = name
|
||||
self.fake_sort = sort
|
||||
|
||||
def do_get_column(self, column):
|
||||
if column == "name":
|
||||
return FakeTagCursor(column, self.fake_name)
|
||||
return FakeTagCursor(column, self.fake_sort)
|
||||
|
||||
class TestTag(unittest.TestCase):
|
||||
def test_tag(self):
|
||||
a = FakeTag(1, "A", "a")
|
||||
b = FakeTag(1, "B", "b")
|
||||
|
||||
self.assertIsInstance(a, objects.Tag)
|
||||
self.assertEqual(a.name, "A")
|
||||
self.assertEqual(a.sort, "a")
|
||||
self.assertEqual(str(a), "A")
|
||||
|
||||
self.assertEqual(a.get_property("name"), "A")
|
||||
self.assertEqual(a.get_property("sort"), "a")
|
||||
|
||||
self.assertTrue(a < b)
|
||||
self.assertTrue(b > a)
|
||||
|
||||
self.assertFalse(a > b)
|
||||
self.assertFalse(b < a)
|
||||
|
||||
|
||||
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_get_all(self): 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_get_all(None)
|
||||
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.get_all(), [ objects.Row(2), objects.Row(3) ])
|
||||
|
||||
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))
|
||||
|
||||
|
||||
class FakeMap(objects.Map):
|
||||
def do_create(self): self.create_called = True
|
||||
def do_insert(self, lhs, rhs): return FakeCursor(1)
|
||||
def do_delete(self, lhs, rhs): return FakeCursor(1)
|
||||
def do_lookup_rhs(self, lhs): return FakeCursor(1)
|
||||
def do_lookup_lhs(self, lhs): return FakeCursor(1)
|
||||
|
||||
class TestMap(unittest.TestCase):
|
||||
def on_row_inserted(self, table, lhs, rhs):
|
||||
self.row_inserted = (lhs, rhs)
|
||||
|
||||
def on_row_deleted(self, table, lhs, rhs):
|
||||
self.row_deleted = (lhs, rhs)
|
||||
|
||||
def test_map_errors(self):
|
||||
with self.assertRaises(NotImplementedError):
|
||||
map = objects.Map("test_map", objects.Tag, objects.Row)
|
||||
with self.assertRaises(NotImplementedError):
|
||||
objects.Map.do_insert(None, True, False)
|
||||
with self.assertRaises(NotImplementedError):
|
||||
objects.Map.do_delete(None, True, False)
|
||||
with self.assertRaises(NotImplementedError):
|
||||
objects.Map.do_lookup_rhs(None, None)
|
||||
with self.assertRaises(NotImplementedError):
|
||||
objects.Map.do_lookup_lhs(None, None)
|
||||
|
||||
def test_map(self):
|
||||
map = FakeMap("test_map", objects.Row, objects.Row)
|
||||
map.connect("row-inserted", self.on_row_inserted)
|
||||
map.connect("row-deleted", self.on_row_deleted)
|
||||
|
||||
self.assertIsInstance(map, GObject.GObject)
|
||||
self.assertEqual(map.map_name, "test_map")
|
||||
self.assertEqual(map.map_lhs, objects.Row)
|
||||
self.assertEqual(map.map_rhs, objects.Row)
|
||||
self.assertTrue(map.create_called)
|
||||
|
||||
map.insert(objects.Row(1), objects.Row(2))
|
||||
self.assertEqual(self.row_inserted, (objects.Row(1), objects.Row(2)))
|
||||
|
||||
map.delete(objects.Row(1), objects.Row(2))
|
||||
self.assertEqual(self.row_deleted, (objects.Row(1), objects.Row(2)))
|
||||
|
||||
rhs_list = map.lookup_rhs(objects.Row(1))
|
||||
self.assertEqual(rhs_list, [ objects.Row(2), objects.Row(3) ])
|
||||
|
||||
lhs_list = map.lookup_lhs(objects.Row(2))
|
||||
self.assertEqual(lhs_list, [ objects.Row(2), objects.Row(3) ])
|
||||
|
|
@ -10,7 +10,6 @@
|
|||
# | name -> playlistid |
|
||||
# +--------------------+
|
||||
from gi.repository import GObject
|
||||
from . import objects
|
||||
from . import playlist
|
||||
from . import sql
|
||||
from . import state
|
||||
|
|
|
|||
Loading…
Reference in New Issue