1
 2import uuid, hashlib, sqlite3
 3import Constants as c
 4
 5def hash(password: str):
 6    salt = uuid.uuid4().hex
 7    return hashlib.sha512(password + salt).hexdigest(), salt
 8
 9def db_query(query, parameters):
10    with sqlite3.connect(c.DB_PATH) as db_connection:
11        curs = db_connection.cursor()
12        curs.execute(query, parameters)
13        return curs
14
15def db_query_one(query, parameters):
16        curs = db_query(query, parameters)
17        try:
18            return list(curs.fetchone())
19        except TypeError:
20            return None
21        
22def run_sql(sql_path: str):
23    with open(sql_path, 'r', encoding="utf-8") as f:
24        with sqlite3.connect(c.DB_PATH) as con:
25            curs = con.cursor()
26            sql = f.read()
27            curs.executescript(sql)
28
29def add_user_to_queue(username, password, email):
30    res = get_user_from_username(username)
31    if res is not None:
32        return None
33    query_str = "insert or ignore into queue (username, password, email) values (?,?,?);"
34    query_param = [username, password, email]
35    return db_query(query_str, query_param)
36
37def add_user(username, password, email):
38    query_str = "insert or ignore into login (username, password, email) values (?,?,?);"
39    query_param = [username, password, email]
40    return db_query(query_str, query_param)
41
42def accept_user(username):
43    r = get_user_from_username(username, "queue")
44    return add_user(r[0], r[1], r[2])
45
46def get_user_from_username(username: str, table="login"):
47    return db_query_one(f"SELECT * FROM { table } where username = (?)", [username])
48
49def get_user_from_id(id: int):
50    return db_query_one("SELECT * from login where user_id = (?)", [id])
51
52def init_db():
53    run_sql(c.MIGRATIONS_INIT_PATH)
54    add_user(c.DEFAULT_ADMIN_USER, c.DEFAULT_ADMIN_PW, c.DEFAULT_ADMIN_EMAIL)