126 lines
3.6 KiB
Python
126 lines
3.6 KiB
Python
# 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
|
|
|
|
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=(Row,Row))
|
|
def row_inserted(self, lhs, rhs): pass
|
|
|
|
@GObject.Signal(arg_types=(Row,Row))
|
|
def row_deleted(self, lhs, rhs): pass
|