@ -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 , db path: str ) :
def __init__ ( self , db _ path: str ) :
"""
: param db path: Path to the database we want to open
: type db path: 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 , q q : 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 q q : Query to execute
: type q q : 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 ( q q )
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 get C ursor( self ) - > sqlite3 :
def get _c ursor( 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 safe query( 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 q q : Query to execute
: type q q : 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 ( q q , deliver )
self . cursor . execute ( q _str , deliver )
self . last_result = self . cursor . fetchall ( )
self . connection . commit ( )
except TypeError as :
print ( " Types in given tuple doesnt match to execute query \" %s \" : %s " % ( q q, e ) , file = stderr )
except TypeError as typ e_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 . safe query( qq , deliver )
except sqlite3 . Error as :
print ( " Couldn ' t execute query %s , exception: %s " % ( q q, e ) , file = stderr )
return self . safe _query( q_str , deliver )
except sqlite3 . Error as sql_qu ery_error :
print ( " Couldn ' t execute query %s , exception: %s " % ( q _str, sql_query_error ) , file = stderr )
print ( deliver )
print ( type ( ) )
print ( type ( sql_qu ery_error ) )
self . last_result = [ ]
return self . last_result
def removeApplicantFromDB ( self , user id: int ) - > bool :
def removeApplicantFromDB ( self , user _ id: int ) - > bool :
""" Removes Applicants from the DB by ID. Use along System.removeUser()
: param user id: User ID to remove from the Database
: type user id: 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 :
print ( f " Could not delete user with id: { user id} , exception in DB: { } " ) # @TODO LOGGING FFS
except sqlite3 . Error as qu ery_error :
print ( f " Could not delete user with id: { user _ id} , exception in DB: { qu ery_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 :
print ( f " Could not delete user { username } , exception in DB: { } " ) # @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 :
print ( f " The database probably doesn ' t exist yet, but read the message: { } " )
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! " )
Darksider3