ssh-reg/private/lib/sqlitedb.py

#!/usr/bin/env python3
import sqlite3
from sys import stderr as STDERR


# create dictionary out of sqlite results
def dict_factory(cursor, row):
    d = {}
    for idx, col in enumerate(cursor.description):
        d[col[0]] = row[idx]
    return d


class SQLitedb:
    """SQLitedb handles EVERYTHING directly related to our Database."""

    db = ""
    cursor = None
    connection = None
    last_result = None

    def __init__(self, dbpath: str):
        """
        :param dbpath: Path to the database we want to open
        :type dbpath: str
        :returns: Object for the SQLitedb-Class.
        :rtype: object
        """

        db = dbpath
        try:
            self.connection = sqlite3.connect(db)
            self.cursor = self.connection.cursor()
        except sqlite3.Error as e:
            print("Connection error: %s" % e, file=STDERR)

        self.cursor.row_factory = dict_factory # every result will be a dict now

    def __del__(self):
        try:
            self.connection.commit()
            self.connection.close()
        except sqlite3.Error as e:
            print("Couldn't gracefully close db: %s" % e, file=STDERR)

    def query(self, qq: str) -> list:
        """Do a query and automagically get the fetched results in a list
        :param qq: Query to execute
        :type qq: str
        :returns: A tuple(/list) consisting with any fetched results
        :rtype: list
        """

        try:
            self.cursor.execute(qq)
            self.last_result = self.cursor.fetchall()
        except sqlite3.Error as e:
            print("Couldn't execute query %s, exception: %s" % (qq, e), 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:
        """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:
        """ 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 deliver: User inputs marked with the placeholder(`?`) in the str
        :type deliver: tuple
        :returns: A tuple(/list) consisting with any fetched results
        :rtype: list
        """

        try:
            self.cursor.execute(qq, deliver)
            self.last_result = self.cursor.fetchall()
        except sqlite3.Error as e:
            print("Couldn't execute query %s, exception: %s" % (qq, e), file=STDERR)
            self.last_result = []
        except TypeError as e:
            print("Types in given tuple doesnt match to execute query \"%s\": %s" % (qq, e), file=STDERR)
            self.last_result = []
        return self.last_result

    def removeApplicantFromDB(self, userid: 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
        :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.connection.commit()
        except sqlite3.Error as e:
            print(f"Could not delete user with id: {userid}, exception in DB: {e}")  # @TODO LOGGING FFS
            return False
        return True

    def removeApplicantFromDBperUsername(self, username: str) -> bool:
        """Removes Applicants from the DB by Username. Use along System.removeUser()
        :param username: Username to remove from the database
        :type username: str
        :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 username = ?", [username])
            self.connection.commit()
        except sqlite3.Error as e:
            print(f"Could not delete user {username}, exception in DB: {e}")  # @TODO LOGGING
            return False
        return True


if __name__ == "__main__":
    try:
        SQLitedb("bla.db")
        print("hi")
        exit(0)
    except KeyboardInterrupt:
        pass