aswi2021jmsd-gitlab/src/dao/private_key_repository.py @ 210c9808

from sqlite3 import Connection, Error, DatabaseError, IntegrityError, ProgrammingError, OperationalError, NotSupportedError
from typing import List

from injector import inject

from src.exceptions.database_exception import DatabaseException
from src.model.private_key import PrivateKey
from src.constants import *
from src.utils.logger import Logger


class PrivateKeyRepository:

    @inject
    def __init__(self, connection: Connection):
        """
        Constructor of the PrivateKeyRepository object

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

        self.connection = connection
        self.cursor = connection.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 (IntegrityError, OperationalError, ProgrammingError, NotSupportedError, DatabaseError, Error) as e:
            Logger.error(str(e))
            raise DatabaseException(e)

        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()

            if private_key_row is None:
                return None

            private_key: PrivateKey = PrivateKey(private_key_row[0],
                                                 private_key_row[1],
                                                 private_key_row[2])

        except (IntegrityError, OperationalError, ProgrammingError, NotSupportedError, DatabaseError, Error) as e:
            Logger.error(str(e))
            raise DatabaseException(e)

        return private_key

    def find_pk(self, private_key_pem: str):
        """
        Tries to find an existing private key by its PEM.
        :param private_key_pem: target key PEM
        :return: corresponding PrivateKey or None if not found
        """

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

            if private_key_row is None:
                return None

            private_key: PrivateKey = PrivateKey(private_key_row[0],
                                                 private_key_row[1],
                                                 private_key_row[2])
        except IntegrityError:
            Logger.error(INTEGRITY_ERROR_MSG)
            raise DatabaseException(INTEGRITY_ERROR_MSG)
        except ProgrammingError:
            Logger.error(PROGRAMMING_ERROR_MSG)
            raise DatabaseException(PROGRAMMING_ERROR_MSG)
        except OperationalError:
            Logger.error(OPERATIONAL_ERROR_MSG)
            raise DatabaseException(OPERATIONAL_ERROR_MSG)
        except NotSupportedError:
            Logger.error(NOT_SUPPORTED_ERROR_MSG)
            raise DatabaseException(NOT_SUPPORTED_ERROR_MSG)
        except DatabaseError:
            Logger.error(DATABASE_ERROR_MSG)
            raise DatabaseException(DATABASE_ERROR_MSG)
        except Error:
            Logger.error(ERROR_MSG)
            raise DatabaseException(ERROR_MSG)

        return private_key

    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 (IntegrityError, OperationalError, ProgrammingError, NotSupportedError, DatabaseError, Error) as e:
            Logger.error(str(e))
            raise DatabaseException(e)

        return private_keys

    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 (IntegrityError, OperationalError, ProgrammingError, NotSupportedError, DatabaseError, Error) as e:
            Logger.error(str(e))
            raise DatabaseException(e)

        return self.cursor.rowcount > 0

    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 (IntegrityError, OperationalError, ProgrammingError, NotSupportedError, DatabaseError, Error) as e:
            Logger.error(str(e))
            raise DatabaseException(e)

        return self.cursor.rowcount > 0