119 lines
4.5 KiB
Python
119 lines
4.5 KiB
Python
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()
|