125 lines
3.5 KiB
Python
125 lines
3.5 KiB
Python
import sqlite3
|
|
|
|
class DatabaseHelper:
|
|
def __init__(self, db=None):
|
|
print("Initializing database")
|
|
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:
|
|
print(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:
|
|
print(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()
|
|
|
|
return True
|
|
|
|
except sqlite3.Error as error:
|
|
print(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:
|
|
print(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:
|
|
print(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:
|
|
print(error)
|
|
return None
|
|
|
|
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 )")
|