Created database class.
Tidied app.py so that it only contains controller logic. Added error handling for if number of entrants exceeds number of possible identifiers Updated test coverage to test for all identifiers, as well as error case
This commit is contained in:
92
web/db.py
Normal file
92
web/db.py
Normal file
@@ -0,0 +1,92 @@
|
||||
import sqlite3
|
||||
import string
|
||||
import utils
|
||||
|
||||
|
||||
class Database:
|
||||
def __init__(self, db_path):
|
||||
if not db_path:
|
||||
raise Exception("DB Path not defined")
|
||||
self.db_path = db_path
|
||||
self.identifiers = self.get_identifiers_list()
|
||||
self.setup_database_tables()
|
||||
|
||||
@staticmethod
|
||||
def get_identifiers_list():
|
||||
"""Returns a list of non ambiguous identifiers"""
|
||||
|
||||
numbers = list(range(1, 100))
|
||||
identifiers = list(string.ascii_uppercase) + [str(item) for item in
|
||||
numbers] # All identifiers are treated as strings
|
||||
ambiguous = ['I', 'O', 'V', '1', '5', '9', '99']
|
||||
utils.remove_common_elements(identifiers, ambiguous)
|
||||
return identifiers
|
||||
|
||||
def get_connection(self):
|
||||
"""Returns sqlite db connection when provided with base directory"""
|
||||
return sqlite3.connect(self.db_path + '/hbc.db')
|
||||
|
||||
def setup_database_tables(self):
|
||||
"""Creates sqlite database and set up the sqlite table if it doesnt already exist"""
|
||||
conn = self.get_connection()
|
||||
|
||||
sql_create_table = """ CREATE TABLE IF NOT EXISTS brewers (
|
||||
id integer PRIMARY KEY,
|
||||
name text NOT NULL,
|
||||
identifier text NOT NULL );"""
|
||||
c = conn.cursor()
|
||||
c.execute(sql_create_table)
|
||||
conn.close()
|
||||
|
||||
def get_identifier(self, name):
|
||||
"""Returns the next availible identifier, passing the result through record_entry to make sure it is not reused"""
|
||||
conn = self.get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute('''select identifier from brewers ORDER BY id DESC LIMIT 1;''')
|
||||
identifier_search_result = c.fetchone()
|
||||
conn.close()
|
||||
|
||||
if identifier_search_result is None:
|
||||
return self.record_entry(self.identifiers[0], name)
|
||||
else:
|
||||
if identifier_search_result[0] == self.identifiers[-1]:
|
||||
raise StopIteration
|
||||
else:
|
||||
i = self.identifiers.index(identifier_search_result[0])
|
||||
return self.record_entry(self.identifiers[i + 1], name)
|
||||
|
||||
def record_entry(self, identifier, name):
|
||||
"""Returns identifier after recording entry in sqlite database"""
|
||||
conn = self.get_connection()
|
||||
c = conn.cursor()
|
||||
c.execute("INSERT INTO brewers (name,identifier) VALUES(?, ?)", (name, identifier))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
return identifier
|
||||
|
||||
|
||||
|
||||
# def get_identifier(name, db_path):
|
||||
# conn = get_connection(db_path)
|
||||
# c = conn.cursor()
|
||||
# c.execute('''select identifier from brewers ORDER BY id DESC LIMIT 1;''')
|
||||
# identifier_search_result = c.fetchone()
|
||||
# conn.close()
|
||||
#
|
||||
# if identifier_search_result is None:
|
||||
# return record_entry(DataStore.identifiers[0], name, db_path)
|
||||
# else:
|
||||
# if identifier_search_result[0] == DataStore.identifiers[-1]:
|
||||
# raise StopIteration
|
||||
# else:
|
||||
# i = DataStore.identifiers.index(identifier_search_result[0])
|
||||
# return record_entry(DataStore.identifiers[i+1], name, db_path)
|
||||
|
||||
|
||||
# def record_entry(identifier, name, db_path):
|
||||
# conn = get_connection(db_path)
|
||||
# c = conn.cursor()
|
||||
# c.execute("INSERT INTO brewers (name,identifier) VALUES(?, ?)", (name, identifier))
|
||||
# conn.commit()
|
||||
# conn.close()
|
||||
# return identifier
|
||||
Reference in New Issue
Block a user