From d2f99d95d06fa7adfca75cf2074fdbf8485b90b0 Mon Sep 17 00:00:00 2001 From: Darksider3 Date: Sat, 30 May 2020 13:48:44 +0200 Subject: [PATCH] sqlitedb.py: Refactor naming safequery->safe_query, more snake_case all over the place enforced. --- private/editUsers.py | 20 +++++------ private/import.py | 2 +- private/lib/ListUsers.py | 2 +- private/lib/Validator.py | 2 +- private/lib/sqlitedb.py | 76 +++++++++++++++++++++------------------- 5 files changed, 53 insertions(+), 49 deletions(-) diff --git a/private/editUsers.py b/private/editUsers.py index 19cf856..9b38e77 100755 --- a/private/editUsers.py +++ b/private/editUsers.py @@ -50,7 +50,7 @@ if __name__ == "__main__": if not DB: print("Could not establish connection to database") 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 if args.remove: @@ -75,9 +75,9 @@ if __name__ == "__main__": print(f"Pubkey '{args.sshpubkey}' isn't valid.") exit(1) try: - DB.safequery("UPDATE `applications` SET `pubkey`=? WHERE `username`=?", - tuple([args.sshpubkey, args.user])) - CurrentUser = DB.safequery("SELECT * FROM `applications` WHERE `username` = ? ", tuple([args.user]))[0] + DB.safe_query("UPDATE `applications` SET `pubkey`=? WHERE `username`=?", + tuple([args.sshpubkey, args.user])) + CurrentUser = DB.safe_query("SELECT * FROM `applications` WHERE `username` = ? ", tuple([args.user]))[0] if int(CurrentUser["status"]) == 1: sys_ctl.make_ssh_usable(args.sshpubkey) except sqlite3.Error as e: @@ -93,7 +93,7 @@ if __name__ == "__main__": print(f"'{args.name}' is not a valid Name.") exit(1) 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: print(f"Could not write '{args.name}' to database: {e}") 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!") exit(1) 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: print(f"Could not write '{args.email}' to the database. {e}") 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: try: - DB.safequery("UPDATE `applications` SET `status` =? WHERE `id`=?", - tuple([args.status, CurrentUser["id"]])) + DB.safe_query("UPDATE `applications` SET `status` =? WHERE `id`=?", + tuple([args.status, CurrentUser["id"]])) sys_ctl.remove_user() except sqlite3.Error as e: 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: try: - DB.safequery("UPDATE `applications` SET `status`=? WHERE `username`=?", - tuple([args.status, args.user])) + DB.safe_query("UPDATE `applications` SET `status`=? WHERE `username`=?", + tuple([args.status, args.user])) sys_ctl.aio_approve(CurrentUser["pubkey"]) except sqlite3.Error as e: print(f"Could not update Users status in database") diff --git a/private/import.py b/private/import.py index 4493078..634132a 100755 --- a/private/import.py +++ b/private/import.py @@ -54,7 +54,7 @@ def import_from_file(file_path: str, database_file: str, user_ids: tuple = tuple elif row["status"] == "0": print(row['username'] + " not approved, therefore not registered.") try: - sql.safequery( + sql.safe_query( "INSERT INTO `applications` (username, name, timestamp, email, pubkey, status) " "VALUES (?,?,?,?,?,?)", tuple([row["username"], row["name"], row["timestamp"], row["email"], row["pubkey"], row["status"]])) diff --git a/private/lib/ListUsers.py b/private/lib/ListUsers.py index 9afec94..88b0a6a 100644 --- a/private/lib/ListUsers.py +++ b/private/lib/ListUsers.py @@ -32,7 +32,7 @@ class ListUsers: self.users_fetch = self.db.query(query) if single_user is not None: 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: """Generates a string with one (approved) single_user per line and one newline at the end diff --git a/private/lib/Validator.py b/private/lib/Validator.py index 154e7d8..a6cb0ea 100644 --- a/private/lib/Validator.py +++ b/private/lib/Validator.py @@ -72,7 +72,7 @@ def checkUserInDB(username: str, db: str) -> bool: try: 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: return True except lib.sqlitedb.sqlite3.Error as e: diff --git a/private/lib/sqlitedb.py b/private/lib/sqlitedb.py index d131e00..3eac2be 100644 --- a/private/lib/sqlitedb.py +++ b/private/lib/sqlitedb.py @@ -1,4 +1,8 @@ #!/usr/bin/env python3 +""" +SQLite wrapper which does just some simple wraps, to ease our experience a little. +""" + import sqlite3 from sys import stderr as stderr from typing import List # Typing support! @@ -20,20 +24,18 @@ class SQLiteDB: connection = 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 - :type dbpath: str + :param db_path: Path to the database we want to open + :type db_path: str :returns: Object for the SQLitedb-Class. :rtype: object """ - - db = dbpath try: - self.connection = sqlite3.connect(db) + self.connection = sqlite3.connect(db_path) self.cursor = self.connection.cursor() - except sqlite3.Error as e: - print("Connection error: %s" % e, file=stderr) + except sqlite3.Error as sql_con: + print("Connection error: %s" % sql_con, file=stderr) self.cursor.row_factory = sqlite3.Row # every result will be a dict now @@ -44,38 +46,39 @@ class SQLiteDB: except sqlite3.Error as e: 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 - :param qq: Query to execute - :type qq: str + :param q_str: Query to execute + :type q_str: str :returns: A tuple(/list) consisting with any fetched results :rtype: list """ try: - self.cursor.execute(qq) + self.cursor.execute(q_str) self.last_result = self.cursor.fetchall() self.connection.commit() except sqlite3.OperationalError: self._createTable() - return self.query(qq) - except sqlite3.Error as e: - print("Couldn't execute query %s, exception: %s" % (qq, e), file=stderr) + return self.query(q_str) + except sqlite3.Error as sql_query_except: + print("Couldn't execute query %s, exception: %s" % (q_str, sql_query_except), + file=stderr) self.last_result = [] return self.last_result # 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 - def getCursor(self) -> sqlite3: + def get_cursor(self) -> sqlite3: """Returns SQLite3 Cursor. Use with **c a u t i o n**... """ 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 # 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 - :param qq: Query to execute - :type qq: str + :param q_str: Query to execute + :type q_str: str :param deliver: User inputs marked with the placeholder(`?`) in the str :type deliver: tuple :returns: A tuple(/list) consisting with any fetched results @@ -83,39 +86,40 @@ class SQLiteDB: """ try: - self.cursor.execute(qq, deliver) + self.cursor.execute(q_str, deliver) self.last_result = self.cursor.fetchall() self.connection.commit() - except TypeError as e: - print("Types in given tuple doesnt match to execute query \"%s\": %s" % (qq, e), file=stderr) + except TypeError as type_err: + print("Types in given tuple doesnt match to execute query \"%s\": %s" % (q_str, type_err), file=stderr) self.last_result = [] except sqlite3.OperationalError: self._createTable() - return self.safequery(qq, deliver) - except sqlite3.Error as e: - print("Couldn't execute query %s, exception: %s" % (qq, e), file=stderr) + return self.safe_query(q_str, deliver) + except sqlite3.Error as sql_query_error: + print("Couldn't execute query %s, exception: %s" % (q_str, sql_query_error), file=stderr) print(deliver) - print(type(e)) + print(type(sql_query_error)) 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() - :param userid: User ID to remove from the Database - :type userid: int + :param user_id: User ID to remove from the Database + :type user_id: int :returns: True, if removal was successful(from the DB), False when not :rtype: bool """ 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() except sqlite3.OperationalError: 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") return False - except sqlite3.Error as e: - print(f"Could not delete user with id: {userid}, exception in DB: {e}") # @TODO LOGGING FFS + except sqlite3.Error as query_error: + print(f"Could not delete user with id: {user_id}, exception in DB: {query_error}") # @TODO LOGGING FFS return False 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" "when a user tried to register") return False - except sqlite3.Error as e: - print(f"Could not delete user {username}, exception in DB: {e}") # @TODO LOGGING + except sqlite3.Error as sql_error: + print(f"Could not delete user {username}, exception in DB: {sql_error}") # @TODO LOGGING return False return True @@ -150,8 +154,8 @@ class SQLiteDB: "timestamp_valid CHECK( timestamp IS strftime('%Y-%m-%d %H:%M:%S', timestamp))" ",status INTEGER NOT NULL DEFAULT 0);") self.connection.commit() - except sqlite3.Error as e: - print(f"The database probably doesn't exist yet, but read the message: {e}") + except sqlite3.Error as sql_error: + 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!")