aswi2021jmsd-gitlab4/src/dao/private_key_repository.py @ e1c31cac

from sqlite3 import Connection, Cursor, Error
from typing import List

from src.model.private_key import PrivateKey
from src.constants import *


class PrivateKeyRepository:

    def __init__(self, connection: Connection, cursor: Cursor):
        """
        Constructor of the PrivateKeyRepository object

        :param connection: Instance of the Connection object
        :param cursor: Instance of the Cursor object
        """

        self.connection = connection
        self.cursor = cursor

    def create(self, private_key: PrivateKey):
        """
        Creates a private key.

        :param private_key: Instance of the PrivateKey object

        :return: the result of whether the creation was successful
        """

        try:
            sql = (f"INSERT INTO {TAB_PRIVATE_KEYS} "
                   f"({COL_PRIVATE_KEY},"
                   f"{COL_PASSWORD}) "
                   f"VALUES(?,?)")
            values = [private_key.private_key,
                      private_key.password]
            self.cursor.execute(sql, values)
            last_id = self.cursor.lastrowid
            self.connection.commit()
        except Error as e:
            print(e)
            return None

        return last_id

    def read(self, private_key_id: int):
        """
        Reads (selects) a private key.

        :param private_key_id: ID of specific private key

        :return: instance of the PrivateKey object
        """

        try:
            sql = (f"SELECT * FROM {TAB_PRIVATE_KEYS} "
                   f"WHERE {COL_ID} = ?")
            values = [private_key_id]
            self.cursor.execute(sql, values)
            private_key_row = self.cursor.fetchone()

            private_key: PrivateKey = PrivateKey(private_key_row[0],
                                                 private_key_row[1],
                                                 private_key_row[2])
        except Error as e:
            print(e)
            return None

        if len(private_key_row) > 0:
            return private_key
        else:
            return None

    def read_all(self):
        """
        Reads (selects) all private keys.

        :return: list of private keys
        """

        try:
            sql = f"SELECT * FROM {TAB_PRIVATE_KEYS}"
            self.cursor.execute(sql)
            private_key_rows = self.cursor.fetchall()

            private_keys: List[PrivateKey] = []
            for private_key_row in private_key_rows:
                private_keys.append(PrivateKey(private_key_row[0],
                                               private_key_row[1],
                                               private_key_row[2]))
        except Error as e:
            print(e)
            return None

        if len(private_keys) > 0:
            return private_keys
        else:
            return None

    def update(self, private_key_id: int, private_key: PrivateKey) -> bool:
        """
        Updates a private key.

        :param private_key_id: ID of specific private key
        :param private_key: Instance of the PrivateKey object

        :return: the result of whether the updation was successful
        """

        try:
            sql = (f"UPDATE {TAB_PRIVATE_KEYS} "
                   f"SET {COL_PRIVATE_KEY} = ?, "
                   f"{COL_PASSWORD} = ? "
                   f"WHERE {COL_ID} = ?")
            values = [private_key.private_key,
                      private_key.password,
                      private_key_id]
            self.cursor.execute(sql, values)
            self.connection.commit()
        except Error as e:
            print(e)
            return False

        return True

    def delete(self, private_key_id: int) -> bool:
        """
        Deletes a private key

        :param private_key_id: ID of specific private key

        :return: the result of whether the deletion was successful
        """

        try:
            sql = (f"DELETE FROM {TAB_PRIVATE_KEYS} "
                   f"WHERE {COL_ID} = ?")
            values = [private_key_id]
            self.cursor.execute(sql, values)
            self.connection.commit()
        except Error as e:
            print(e)
            return False

        return self.cursor.rowcount > 0