aswi2021jmsd-gitlab/src/dao/private_key_repository.py @ bcdb15bd

from ..db_objects.private_key import PrivateKey
from ..dao.repository import IRepository
from db_manager import DBManager
from ..constants import *


class PrivateKeyRepositoryImpl(IRepository):

    def create(self, private_key: str, password: str) -> bool:
        """
        Creates a private key

        :param private_key: private key
        :param password: passphrase for encode private key

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

        sql = f"INSERT INTO {TAB_PRIVATE_KEYS} ({COL_PRIVATE_KEY},{COL_PASSWORD}) VALUES(?,?)"
        result = DBManager.create(sql, [private_key, password])
        return result

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

        :param private_key_id: ID of specific private

        :return: instance of PrivateKey class if ID was used otherwise list of this instances
        """

        sql_extend = ""

        if private_key_id is not None:
            sql_extend = f" WHERE {COL_ID} = ?"

        sql = f"SELECT * FROM {TAB_PRIVATE_KEYS}{sql_extend}"
        rows = DBManager.read(sql, private_key_id)

        private_keys: list = []

        for row in rows:
            private_keys.append(PrivateKey(row))

        if private_key_id is not None:
            return private_keys[0]

        return private_keys

    def update(self, private_key_id: int, private_key: str = None, password: str = None) -> bool:
        """
        Updates a private key

        :param private_key_id: ID of specific private key
        :param private_key: private key
        :param password: passphrase for encode private key

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

        updated_list = []
        values = []
        if private_key is not None:
            updated_list.append(f"{COL_PRIVATE_KEY} = ?")
            values.append(private_key)
        if password is not None:
            updated_list.append(f"{COL_PASSWORD} = ?")
            values.append(password)

        updated_str = ", ".join(updated_list)
        sql = f"UPDATE {TAB_PRIVATE_KEYS} SET {updated_str} WHERE {COL_ID} = ?"
        result = DBManager.update(sql, private_key_id, values)
        return result

    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
        """

        sql = f"DELETE FROM {TAB_PRIVATE_KEYS} WHERE {COL_ID} = ?"
        result = DBManager.delete(sql, private_key_id, False)
        return result