Revize 9d3eded8

Re #8471 - Simplification of CRUD functions
- conversion source of private key repository to one file
- added comments

src/dao/private_key_repository.py

@@ -1 @@
-from ..db_objects.private_key import PrivateKey
-from ..dao.repository import IRepository
-from db_manager import DBManager
-from ..constants import *
+from sqlite3 import Connection, Cursor, Error
 
+from src.db_objects.private_key import PrivateKey
+from src.constants import *
 
-class PrivateKeyRepositoryImpl(IRepository):
 
-    def create(self, private_key: str, password: str) -> bool:
-        """
-        Creates a private key
+class PrivateKeyRepository:
 
-        :param private_key: private key
-        :param password: passphrase for encode private key
+    def __init__(self, connection: Connection, cursor: Cursor):
+        """
+        Constructor of the PrivateKeyRepository object
 
-        :return: the result of whether the creation was successful
+        :param connection: Instance of the Connection object
+        :param cursor: Instance of the Cursor object
         """
 
-        sql = f"INSERT INTO {TAB_PRIVATE_KEYS} ({COL_PRIVATE_KEY},{COL_PASSWORD}) VALUES(?,?)"
-        result = DBManager.create(sql, [private_key, password])
-        return result
+        self.connection = connection
+        self.cursor = cursor
 
-    def read(self, private_key_id: int = None):
+    def create(self, private_key: PrivateKey) -> bool:
         """
-        Reads (selects) a private key
+        Creates a private key.
 
-        :param private_key_id: ID of specific private
+        :param private_key: Instance of the PrivateKey object
 
-        :return: instance of PrivateKey class if ID was used otherwise list of this instances
+        :return: the result of whether the creation was successful
         """
 
-        sql_extend = ""
+        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)
+            self.connection.commit()
+
+        except Error as e:
+            print(e)
+            return False
 
-        if private_key_id is not None:
-            sql_extend = f" WHERE {COL_ID} = ?"
+        return True
 
-        sql = f"SELECT * FROM {TAB_PRIVATE_KEYS}{sql_extend}"
-        rows = DBManager.read(sql, private_key_id)
+    def read(self, private_key_id: int):
+        """
+        Reads (selects) a private key.
 
-        private_keys: list = []
+        :param private_key_id: ID of specific private key
 
-        for row in rows:
-            private_keys.append(PrivateKey(row))
+        :return: instance of the PrivateKey object
+        """
 
-        if private_key_id is not None:
-            return private_keys[0]
+        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.fetchall()
 
-        return private_keys
+            private_key: PrivateKey = PrivateKey(private_key_row[0],
+                                                 private_key_row[1],
+                                                 private_key_row[2])
+        except Error as e:
+            print(e)
+            return None
 
-    def update(self, private_key_id: int, private_key: str = None, password: str = None) -> bool:
+        return private_key
+
+    def update(self, private_key_id: int, private_key: PrivateKey) -> bool:
         """
-        Updates a private key
+        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
+        :param private_key: Instance of the PrivateKey object
 
         :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
+        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]
+            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:
         """
@@ -81 +101 @@
         :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
+        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 True