Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- from api.models import models
- from api.validators import validator
- from typing import List
- import sqlite3 as sql
- import os
- import uuid
- class DbEngine:
- def create_db_path(self):
- curr_dir = os.getcwd()
- api_dir = os.path.join(curr_dir, 'api')
- db_dir = os.path.join(api_dir, 'database')
- db_file = os.path.join(db_dir, 'main.db')
- return db_file
- def create_tables(self) -> bool:
- file_path = self.create_path()
- with sql.connect(file_path) as mdb:
- cursor = mdb.cursor()
- try:
- cursor.execute('CREATE TABLE IF NOT EXISTS Users (Id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, Username TEXT, Password TEXT)')
- except Exception as e:
- raise e
- except sql.Exception as err:
- raise err
- try:
- cursor.execute('CREATE TABLE IF NOT EXISTS Passwords (Id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, Username TEXT, Password TEXT)')
- except Exception as e:
- raise e
- except sql.Exception as err:
- raise err
- def query_user_login(self, user: models.User) -> List[bool, str]:
- file_path = self.create_db_path()
- with sql.connect(file_path) as mdb:
- cursor = mdb.cursor()
- try:
- results = cursor.execute(
- 'SELECT Password FROM Users WHERE Username=?',
- (user.username, )
- ).fetchone()
- if not results or results[0]:
- return False
- return True
- except Exception as e:
- raise e
- except sql.Error as err:
- raise err
- def save_user(self, user: models.User):
- file_path = self.create_db_path()
- with sql.connect(file_path) as mdb:
- cursor = mdb.cursor()
- generated_id = uuid.uuid4()
- all_ids = self.get_ids()
- while generated_id in all_ids:
- generated_id = uuid.uuid4()
- try:
- cursor.execute(
- 'INSERT OR IGNORE INTO Users(Id, Username, Password) VALUES (?,?,?)',
- (generated_id, user.username, user.password)
- )
- return True
- except Exception as e:
- raise e
- except sql.Error as err:
- raise err
- def get_user(self, username: str) -> models.User:
- file_path = self.create_db_path()
- with sql.connect(file_path) as mdb:
- cursor = mdb.cursor()
- srch = 'SELECT Id, Password FROM Users WHERE Username=?'
- srch2 = 'SELECT Passwords FROM Passwords WHERE Username=?'
- try:
- user_results = cursor.execute(srch, (username, )).fetchone()
- if not user_results[0]:
- return False
- password_results = cursor.execute(srch2, (username, )).fetchall()
- return_user = models.User(
- id = user_results[0],
- username = username,
- password = user_results[1],
- saved_passwords = []
- )
- for password in password_results:
- return_user.saved_passwords.append(password[0])
- return return_user
- except Exception as e:
- raise e
- except sql.Error as err:
- raise err
Advertisement
Add Comment
Please, Sign In to add comment