server code with basic account stuff

server/db.py

@@ -0,0 +1,118 @@
+import os
+import sqlite3
+
+class Database():
+    def __init__(self,name):
+        print(f"init db {name}")
+        self.name = name
+        self.con = sqlite3.connect(f"{name}.db")
+        self.cur = self.con.cursor()
+
+    def execute(self,cmd: str,*args):
+        print("execute",[cmd,*args])
+        return self.cur.execute(cmd,*args)
+
+    def executemany(self,cmd: str,*args):
+        print("executemany",[cmd,*args])
+        return self.cur.executemany(cmd,*args)
+
+    def select(self,cmd: str,*args) -> list[tuple]:
+        return self.execute(cmd,*args).fetchall()
+
+    def sameType(self,typeA: str,typeB: str) -> bool:
+        return typeA.lower() == typeB.lower()
+
+
+    # tables
+    def tableExists(self,table: str) -> bool:
+        res = self.select("SELECT name FROM sqlite_master WHERE type='table' AND name=?",(table,))
+        return type(res)!=type(None) and len(res)>0
+
+    def createTable(self,table: str,cols: dict[str,str]) -> None:
+        if self.tableExists(table): return self.checkColumns(table,cols)
+        # i know i shouldn't be inserting these directly but using placeholders returns a syntax error for some reason
+        self.execute(f"CREATE TABLE {table}({", ".join([f"{n} {cols[n]}" for n in cols])})")
+
+    def dropTable(self,table: str) -> None:
+        self.execute(f"DROP TABLE {table}")
+
+
+    # columns
+    def getColumns(self,table: str) -> tuple[str]:
+        if not self.tableExists(table): return ()
+        res = self.select(f"PRAGMA table_info('{table}')")
+        return dict([e[1:3] for e in res])
+
+    def hasColumn(self,table: str,colname: str) -> bool:
+        return colname in self.getColumns(table)
+
+    def insertColumn(self,table: str,colname: str,coltype: str) -> None:
+        self.execute(f"ALTER TABLE {table} ADD COLUMN {colname} {coltype}")
+
+    def checkColumns(self,table: str,cols: dict[str,str]) -> None:
+        tcols = self.getColumns(table)
+        for colname in cols:
+            if not colname in tcols:
+                self.insertColumn(table,colname,cols[colname])
+                continue
+            if not self.sameType(cols[colname],tcols[colname]):
+                raise Exception(f"Invalid table '{table}' in database '{self.name}': column '{colname}' has type '{tcols[colname]}' instead of '{cols[colname]}'")
+
+
+    # rows
+    def insertRows(self,table: str,values: list[list|tuple|dict]) -> None:
+        if len(values)==0: return
+        amount=len(values[0])
+        if type(values[0])==dict:
+            cols = self.getColumns(table)
+            for e in values:
+                for c in cols:
+                    if not c in e:
+                        e[c] = None
+            self.executemany(f"INSERT INTO {table} VALUES({", ".join([f":{e}" for e in cols])})",values)
+        else:
+            self.executemany(f"INSERT INTO {table} VALUES({", ".join(["?"]*amount)})",values)
+        self.con.commit()
+
+    def insertRow(self,table: str,values: list|tuple|dict) -> None:
+        self.insertRows(table,[values])
+
+    def rowAmount(self,table: str) -> int:
+        return self.select(f"SELECT COUNT(*) FROM {table}")[0][0]
+
+    # find
+    def findRowList(self,table: str,pattern: str,*args) -> list | None:
+        res = self.select(f"SELECT * FROM {table} WHERE {pattern}",*args)
+        if len(res)==0: return None
+        return res[0]
+
+    def findRow(self,table: str,pattern: str,*args) -> dict | None:
+        res = self.findRowList(table,pattern,*args)
+        if res==None: return None
+        cols = list(self.getColumns(table))
+        return dict([[cols[i],res[i]] for i in range(len(cols))])
+
+    # PK = primary key
+    def rowListFromPK(self,table: str,pkname: str,pk) -> tuple | None:
+        res = self.select(f"SELECT * FROM {table} WHERE {pkname} = ?",(pk,))
+        if len(res)==0: return None
+        return res[0]
+
+    def rowFromPK(self,table: str,pkname: str,pk) -> dict | None:
+        res = self.rowListFromPK(table,pkname,pk)
+        if res==None: return None
+        cols = list(self.getColumns(table))
+        return dict([[cols[i],res[i]] for i in range(len(cols))])
+
+    def hasPK(self,table: str,pkname: str,pk) -> bool:
+        return type(self.rowListFromPK(table,pkname,pk))!=type(None)
+
+    def removePK(self,table: str,pkname: str,pk) -> None:
+        self.execute(f"DELETE FROM {table} WHERE {pkname} = ?",(pk,))
+        self.con.commit()
+
+    def updateFromPK(self,table: str,pkname: str,pk,row:dict):
+        keys = list(row)
+        vals = [row[i] for i in keys]
+        self.execute(f"UPDATE {table} SET {", ".join([f"{e} = ?" for e in keys])} WHERE {pkname} = ?",(*vals,pk))
+        self.con.commit()