dark
/
ssh-reg
forked from tilde/ssh-reg
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

sqlitedb.py: Refactor naming

Browse Source 
safequery->safe_query, more snake_case all over the place enforced.
dev
Darksider3 4 years ago
parent fb620447c7
commit d2f99d95d0
5 changed files with 53 additions and 49 deletions
Show Stats Download Patch File Download Diff File

20
private/editUsers.py
Unescape Escape View File

@ -50,7 +50,7 @@ if __name__ == "__main__":
if not DB: if not DB:
print("Could not establish connection to database") print("Could not establish connection to database")
exit(1) exit(1)
CurrentUser = DB.safequery("SELECT * FROM `applications` WHERE `username`=?", tuple([args.user]))[0] CurrentUser = DB.safe_query("SELECT * FROM `applications` WHERE `username`=?", tuple([args.user]))[0]
# --> --remove # --> --remove
if args.remove: if args.remove:
@ -75,9 +75,9 @@ if __name__ == "__main__":
print(f"Pubkey '{args.sshpubkey}' isn't valid.") print(f"Pubkey '{args.sshpubkey}' isn't valid.")
exit(1) exit(1)
try: try:
DB.safequery("UPDATE `applications` SET `pubkey`=? WHERE `username`=?", DB.safe_query("UPDATE `applications` SET `pubkey`=? WHERE `username`=?",
tuple([args.sshpubkey, args.user])) tuple([args.sshpubkey, args.user]))
CurrentUser = DB.safequery("SELECT * FROM `applications` WHERE `username` = ? ", tuple([args.user]))[0] CurrentUser = DB.safe_query("SELECT * FROM `applications` WHERE `username` = ? ", tuple([args.user]))[0]
if int(CurrentUser["status"]) == 1: if int(CurrentUser["status"]) == 1:
sys_ctl.make_ssh_usable(args.sshpubkey) sys_ctl.make_ssh_usable(args.sshpubkey)
except sqlite3.Error as e: except sqlite3.Error as e:
@ -93,7 +93,7 @@ if __name__ == "__main__":
print(f"'{args.name}' is not a valid Name.") print(f"'{args.name}' is not a valid Name.")
exit(1) exit(1)
try: try:
DB.safequery("UPDATE `applications` SET `name` =? WHERE `username` =?", tuple([args.name, args.user])) DB.safe_query("UPDATE `applications` SET `name` =? WHERE `username` =?", tuple([args.name, args.user]))
except sqlite3.Error as e: except sqlite3.Error as e:
print(f"Could not write '{args.name}' to database: {e}") print(f"Could not write '{args.name}' to database: {e}")
print(f"'{args.user}'s Name changed to '{args.name}'.") print(f"'{args.user}'s Name changed to '{args.name}'.")
@ -104,7 +104,7 @@ if __name__ == "__main__":
print(f"'{args.email}' is not a valid Mail address!") print(f"'{args.email}' is not a valid Mail address!")
exit(1) exit(1)
try: try:
DB.safequery("UPDATE `applications` SET `email` =? WHERE `username` =?", tuple([args.email])) DB.safe_query("UPDATE `applications` SET `email` =? WHERE `username` =?", tuple([args.email]))
except sqlite3.Error as e: except sqlite3.Error as e:
print(f"Could not write '{args.email}' to the database. {e}") print(f"Could not write '{args.email}' to the database. {e}")
print(f"'{args.user}' Mail changed to '{args.email}'.") print(f"'{args.user}' Mail changed to '{args.email}'.")
@ -121,8 +121,8 @@ if __name__ == "__main__":
if args.status == 0 and int(CurrentUser["status"]) == 1: if args.status == 0 and int(CurrentUser["status"]) == 1:
try: try:
DB.safequery("UPDATE `applications` SET `status` =? WHERE `id`=?", DB.safe_query("UPDATE `applications` SET `status` =? WHERE `id`=?",
tuple([args.status, CurrentUser["id"]])) tuple([args.status, CurrentUser["id"]]))
sys_ctl.remove_user() sys_ctl.remove_user()
except sqlite3.Error as e: except sqlite3.Error as e:
print(f"Could not update database entry for '{args.user}', did not touch the system") print(f"Could not update database entry for '{args.user}', did not touch the system")
@ -134,8 +134,8 @@ if __name__ == "__main__":
if args.status == 1 and int(CurrentUser["status"]) == 0: if args.status == 1 and int(CurrentUser["status"]) == 0:
try: try:
DB.safequery("UPDATE `applications` SET `status`=? WHERE `username`=?", DB.safe_query("UPDATE `applications` SET `status`=? WHERE `username`=?",
tuple([args.status, args.user])) tuple([args.status, args.user]))
sys_ctl.aio_approve(CurrentUser["pubkey"]) sys_ctl.aio_approve(CurrentUser["pubkey"])
except sqlite3.Error as e: except sqlite3.Error as e:
print(f"Could not update Users status in database") print(f"Could not update Users status in database")

2
private/import.py
Unescape Escape View File

@ -54,7 +54,7 @@ def import_from_file(file_path: str, database_file: str, user_ids: tuple = tuple
elif row["status"] == "0": elif row["status"] == "0":
print(row['username'] + " not approved, therefore not registered.") print(row['username'] + " not approved, therefore not registered.")
try: try:
sql.safequery( sql.safe_query(
"INSERT INTO `applications` (username, name, timestamp, email, pubkey, status) " "INSERT INTO `applications` (username, name, timestamp, email, pubkey, status) "
"VALUES (?,?,?,?,?,?)", tuple([row["username"], row["name"], row["timestamp"], "VALUES (?,?,?,?,?,?)", tuple([row["username"], row["name"], row["timestamp"],
row["email"], row["pubkey"], row["status"]])) row["email"], row["pubkey"], row["status"]]))

2
private/lib/ListUsers.py
Unescape Escape View File

@ -32,7 +32,7 @@ class ListUsers:
self.users_fetch = self.db.query(query) self.users_fetch = self.db.query(query)
if single_user is not None: if single_user is not None:
query = "SELECT * FROM `applications` WHERE `username` = ?" query = "SELECT * FROM `applications` WHERE `username` = ?"
self.users_fetch = self.db.safequery(query, tuple([single_user])) self.users_fetch = self.db.safe_query(query, tuple([single_user]))
def output_as_list(self) -> str: def output_as_list(self) -> str:
"""Generates a string with one (approved) single_user per line and one newline at the end """Generates a string with one (approved) single_user per line and one newline at the end

2
private/lib/Validator.py
Unescape Escape View File

@ -72,7 +72,7 @@ def checkUserInDB(username: str, db: str) -> bool:
try: try:
ldb = lib.sqlitedb.SQLiteDB(db) ldb = lib.sqlitedb.SQLiteDB(db)
fetched = ldb.safequery("SELECT * FROM 'applications' WHERE username = ?", tuple([username])) fetched = ldb.safe_query("SELECT * FROM 'applications' WHERE username = ?", tuple([username]))
if fetched: if fetched:
return True return True
except lib.sqlitedb.sqlite3.Error as e: except lib.sqlitedb.sqlite3.Error as e:

76
private/lib/sqlitedb.py
Unescape Escape View File

@ -1,4 +1,8 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""
SQLite wrapper which does just some simple wraps, to ease our experience a little.
"""
import sqlite3 import sqlite3
from sys import stderr as stderr from sys import stderr as stderr
from typing import List # Typing support! from typing import List # Typing support!
@ -20,20 +24,18 @@ class SQLiteDB:
connection = None connection = None
last_result = None last_result = None
def __init__(self, dbpath: str): def __init__(self, db_path: str):
""" """
:param dbpath: Path to the database we want to open :param db_path: Path to the database we want to open
:type dbpath: str :type db_path: str
:returns: Object for the SQLitedb-Class. :returns: Object for the SQLitedb-Class.
:rtype: object :rtype: object
""" """
db = dbpath
try: try:
self.connection = sqlite3.connect(db) self.connection = sqlite3.connect(db_path)
self.cursor = self.connection.cursor() self.cursor = self.connection.cursor()
except sqlite3.Error as e: except sqlite3.Error as sql_con:
print("Connection error: %s" % e, file=stderr) print("Connection error: %s" % sql_con, file=stderr)
self.cursor.row_factory = sqlite3.Row # every result will be a dict now self.cursor.row_factory = sqlite3.Row # every result will be a dict now
@ -44,38 +46,39 @@ class SQLiteDB:
except sqlite3.Error as e: except sqlite3.Error as e:
print("Couldn't gracefully close db: %s" % e, file=stderr) print("Couldn't gracefully close db: %s" % e, file=stderr)
def query(self, qq: str) -> List[sqlite3.Row]: def query(self, q_str: str) -> List[sqlite3.Row]:
"""Do a query and automagically get the fetched results in a list """Do a query and automagically get the fetched results in a list
:param qq: Query to execute :param q_str: Query to execute
:type qq: str :type q_str: str
:returns: A tuple(/list) consisting with any fetched results :returns: A tuple(/list) consisting with any fetched results
:rtype: list :rtype: list
""" """
try: try:
self.cursor.execute(qq) self.cursor.execute(q_str)
self.last_result = self.cursor.fetchall() self.last_result = self.cursor.fetchall()
self.connection.commit() self.connection.commit()
except sqlite3.OperationalError: except sqlite3.OperationalError:
self._createTable() self._createTable()
return self.query(qq) return self.query(q_str)
except sqlite3.Error as e: except sqlite3.Error as sql_query_except:
print("Couldn't execute query %s, exception: %s" % (qq, e), file=stderr) print("Couldn't execute query %s, exception: %s" % (q_str, sql_query_except),
file=stderr)
self.last_result = [] self.last_result = []
return self.last_result return self.last_result
# sometimes we need the cursor for safety reasons, for example does sqlite3 all the security related # sometimes we need the cursor for safety reasons, for example does sqlite3 all the security related
# escaoing in supplied strings for us, when we deliver it to con.execute in the second argument as a tuple # escaoing in supplied strings for us, when we deliver it to con.execute in the second argument as a tuple
def getCursor(self) -> sqlite3: def get_cursor(self) -> sqlite3:
"""Returns SQLite3 Cursor. Use with **c a u t i o n**... """ """Returns SQLite3 Cursor. Use with **c a u t i o n**... """
return self.cursor return self.cursor
# we could try to utilise that ourselfs in a function. Be c a r e f u l, these values in the tuple MUST HAVE # we could try to utilise that ourselfs in a function. Be c a r e f u l, these values in the tuple MUST HAVE
# THE RIGHT TYPE # THE RIGHT TYPE
def safequery(self, qq: str, deliver: tuple) -> List[sqlite3.Row]: def safe_query(self, q_str: str, deliver: tuple) -> List[sqlite3.Row]:
""" Shall handle any query that has user input in it as an alternative to self.query """ Shall handle any query that has user input in it as an alternative to self.query
:param qq: Query to execute :param q_str: Query to execute
:type qq: str :type q_str: str
:param deliver: User inputs marked with the placeholder(`?`) in the str :param deliver: User inputs marked with the placeholder(`?`) in the str
:type deliver: tuple :type deliver: tuple
:returns: A tuple(/list) consisting with any fetched results :returns: A tuple(/list) consisting with any fetched results
@ -83,39 +86,40 @@ class SQLiteDB:
""" """
try: try:
self.cursor.execute(qq, deliver) self.cursor.execute(q_str, deliver)
self.last_result = self.cursor.fetchall() self.last_result = self.cursor.fetchall()
self.connection.commit() self.connection.commit()
except TypeError as e: except TypeError as type_err:
print("Types in given tuple doesnt match to execute query \"%s\": %s" % (qq, e), file=stderr) print("Types in given tuple doesnt match to execute query \"%s\": %s" % (q_str, type_err), file=stderr)
self.last_result = [] self.last_result = []
except sqlite3.OperationalError: except sqlite3.OperationalError:
self._createTable() self._createTable()
return self.safequery(qq, deliver) return self.safe_query(q_str, deliver)
except sqlite3.Error as e: except sqlite3.Error as sql_query_error:
print("Couldn't execute query %s, exception: %s" % (qq, e), file=stderr) print("Couldn't execute query %s, exception: %s" % (q_str, sql_query_error), file=stderr)
print(deliver) print(deliver)
print(type(e)) print(type(sql_query_error))
self.last_result = [] self.last_result = []
return self.last_result return self.last_result
def removeApplicantFromDB(self, userid: int) -> bool: def removeApplicantFromDB(self, user_id: int) -> bool:
"""Removes Applicants from the DB by ID. Use along System.removeUser() """Removes Applicants from the DB by ID. Use along System.removeUser()
:param userid: User ID to remove from the Database :param user_id: User ID to remove from the Database
:type userid: int :type user_id: int
:returns: True, if removal was successful(from the DB), False when not :returns: True, if removal was successful(from the DB), False when not
:rtype: bool :rtype: bool
""" """
try: try:
self.last_result = self.cursor.execute("DELETE FROM `applications` WHERE id = ? ", [userid]) self.last_result = self.cursor.execute("DELETE FROM `applications` WHERE id = ? ",
[user_id])
self.connection.commit() self.connection.commit()
except sqlite3.OperationalError: except sqlite3.OperationalError:
print("The database has probably not yet seen any users, so it didnt create your table yet. Come back" print("The database has probably not yet seen any users, so it didnt create your table yet. Come back"
"when a user tried to register") "when a user tried to register")
return False return False
except sqlite3.Error as e: except sqlite3.Error as query_error:
print(f"Could not delete user with id: {userid}, exception in DB: {e}") # @TODO LOGGING FFS print(f"Could not delete user with id: {user_id}, exception in DB: {query_error}") # @TODO LOGGING FFS
return False return False
return True return True
@ -134,8 +138,8 @@ class SQLiteDB:
print("The database has probably not yet seen any users, so it didnt create your table yet. Come back" print("The database has probably not yet seen any users, so it didnt create your table yet. Come back"
"when a user tried to register") "when a user tried to register")
return False return False
except sqlite3.Error as e: except sqlite3.Error as sql_error:
print(f"Could not delete user {username}, exception in DB: {e}") # @TODO LOGGING print(f"Could not delete user {username}, exception in DB: {sql_error}") # @TODO LOGGING
return False return False
return True return True
@ -150,8 +154,8 @@ class SQLiteDB:
"timestamp_valid CHECK( timestamp IS strftime('%Y-%m-%d %H:%M:%S', timestamp))" "timestamp_valid CHECK( timestamp IS strftime('%Y-%m-%d %H:%M:%S', timestamp))"
",status INTEGER NOT NULL DEFAULT 0);") ",status INTEGER NOT NULL DEFAULT 0);")
self.connection.commit() self.connection.commit()
except sqlite3.Error as e: except sqlite3.Error as sql_error:
print(f"The database probably doesn't exist yet, but read the message: {e}") print(f"The database probably doesn't exist yet, but read the message: {sql_error}")
print("The database table didn't exist yet; created it successfully!") print("The database table didn't exist yet; created it successfully!")

Write Preview
Loading…
Cancel
Save