170 lines
4.9 KiB
Python
170 lines
4.9 KiB
Python
import sqlite3
|
|
from helpers import UtilityHelper as utilityHelper
|
|
|
|
class DatabaseHelper:
|
|
def __init__(self, db=None):
|
|
self.conn = None
|
|
self.cursor = None
|
|
|
|
if db:
|
|
self.open(db)
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self):
|
|
self.close()
|
|
|
|
def open(self, db):
|
|
try:
|
|
self.conn = sqlite3.connect(db)
|
|
self.cursor = self.conn.cursor()
|
|
except:
|
|
pass
|
|
|
|
def commit(self):
|
|
if self.conn:
|
|
self.conn.commit()
|
|
|
|
def query(self, sql, values=None):
|
|
if values == None:
|
|
self.cursor.execute(sql)
|
|
self.commit()
|
|
else:
|
|
self.cursor.execute(sql, values)
|
|
self.commit()
|
|
|
|
def close(self):
|
|
if self.conn:
|
|
self.conn.commit()
|
|
self.conn.close()
|
|
self.cursor.close()
|
|
|
|
# Query functions
|
|
def loginUser(self, username, password):
|
|
if not self.conn:
|
|
return None
|
|
try:
|
|
self.cursor.execute("SELECT * FROM `users` WHERE (`username` = ? AND `password` = ?)", (username, password,))
|
|
return self.cursor.fetchone()
|
|
except sqlite3.Error as error:
|
|
return None
|
|
|
|
def createUser(self, private_key, public_key, username, password):
|
|
if not self.conn:
|
|
return None
|
|
try:
|
|
self.cursor.execute("INSERT INTO `users` (private_key, public_key, username, password) VALUES (?, ?, ?, ?)", (private_key, public_key, username, password,))
|
|
self.commit()
|
|
return True
|
|
except sqlite3.Error as error:
|
|
return False
|
|
|
|
def changePassword(self, user_privatekey, old_password, password):
|
|
if not self.conn:
|
|
return None
|
|
|
|
try:
|
|
self.cursor.execute("UPDATE `users` SET `password` = ? WHERE (`private_key` = ? AND `password` = ?)", (password, user_privatekey, old_password,))
|
|
self.commit()
|
|
if self.cursor.rowcount < 1:
|
|
return False
|
|
|
|
return True
|
|
|
|
except sqlite3.Error as error:
|
|
return None
|
|
|
|
def changeUsername(self, user_privatekey, username):
|
|
if not self.conn:
|
|
return None
|
|
|
|
# Execute the query
|
|
try:
|
|
self.cursor.execute("UPDATE `users` SET username = ? WHERE `private_key` = ?", (username, user_privatekey,))
|
|
self.commit()
|
|
|
|
return True
|
|
|
|
except sqlite3.Error as error:
|
|
return None
|
|
|
|
def deleteUser(self, user_privatekey):
|
|
# Check if the database is open
|
|
if not self.conn:
|
|
return None
|
|
|
|
# Execute the query
|
|
try:
|
|
self.cursor.execute("DELETE FROM `users` WHERE `private_key` = ?", (user_privatekey, ))
|
|
self.commit()
|
|
|
|
return True
|
|
|
|
except sqlite3.Error as error:
|
|
return None
|
|
|
|
def fetchUserByUsername(self, username):
|
|
if not self.conn:
|
|
return None
|
|
|
|
try:
|
|
self.cursor.execute("SELECT * FROM `users` WHERE `username` = ?", (username,))
|
|
|
|
return self.cursor.fetchone()
|
|
|
|
except sqlite3.Error as error:
|
|
return None
|
|
|
|
def createLog(self, public_key, date, info, unread = 0):
|
|
if not self.conn:
|
|
return None
|
|
|
|
try:
|
|
self.cursor.execute("INSERT INTO `logs` (public_key, date, info, unread) VALUES (?, ?, ?, ?)", (public_key, date, info, unread,))
|
|
self.commit()
|
|
|
|
except sqlite3.Error as error:
|
|
return None
|
|
|
|
|
|
def getLatestLogs(self):
|
|
if not self.conn:
|
|
return None
|
|
|
|
try:
|
|
self.cursor.execute("SELECT * FROM `logs` ORDER BY `id` DESC LIMIT 10")
|
|
|
|
return self.cursor.fetchall()
|
|
|
|
except sqlite3.Error as error:
|
|
return None
|
|
|
|
def getUreandLogs(self):
|
|
if not self.conn:
|
|
return None
|
|
|
|
try:
|
|
self.cursor.execute("SELECT * FROM `logs` WHERE `unread` = ?", (0,))
|
|
|
|
return self.cursor.fetchall()
|
|
|
|
except sqlite3.Error as error:
|
|
return None
|
|
|
|
|
|
def updateLogStatus(self, id, unread = 1):
|
|
if not self.conn:
|
|
return False
|
|
|
|
try:
|
|
self.cursor.execute("UPDATE `logs` SET `unread` = ? WHERE `id` = ?", (unread, id, ))
|
|
self.commit()
|
|
|
|
except sqlite3.Error as error:
|
|
return False
|
|
|
|
def checkMigrations(self):
|
|
self.query("CREATE TABLE IF NOT EXISTS `users` ( `private_key` BLOB NOT NULL , `public_key` BLOB NOT NULL , `username` VARCHAR(255) NOT NULL , `password` VARCHAR(255) NOT NULL )")
|
|
self.query("CREATE TABLE IF NOT EXISTS `logs` ( `id` INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, `public_key` BLOB NOT NULL , `date` VARCHAR(255) NOT NULL , `info` VARCHAR(255) NOT NULL , `unread` INTEGER NOT NULL )")
|