Revize f8b23532

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

src/dao/certificate_repository.py

@@ -1 @@
-from typing import List, Dict
+from typing import Dict
+from sqlite3 import Connection, Cursor, Error
 
-from ..db_objects.certificate import Certificate
-from ..dao.repository import IRepository
-from db_manager import DBManager
-from ..constants import *
+from src.db_objects.certificate import Certificate
+from src.constants import *
 
 
-class CertificateRepositoryImpl(IRepository):
+class CertificateRepository:
 
-    def create(self, common_name: str, valid_from: str, valid_to: str, pem_data: str,
-               private_key_id: int, type_id: int, parent_id: int, usages: Dict[int, bool]) -> bool:
+    def __init__(self, connection: Connection, cursor: Cursor):
         """
-        Creates a certificate
-
-        :param common_name: private key of the certificate
-        :param valid_from: string of date (time) of validation after (from)
-        :param valid_to: string of date (time) of validation before (to)
-        :param pem_data: data of certificate in a PEM format
-        :param private_key_id: ID from PrivateKeys table for specific private key
-        :param type_id: ID from CertificateTypes table (ROOT_CA_ID=1, INTERMEDIATE_CA_ID=2, CERTIFICATE_ID=3)
-        :param parent_id: ID from Certificates table for specific parent certificate
-            for root CA use e.g. -1 (ID will be changed after created certificate)
-        :param usages: dictionary of usages IDs which certificate use
-            Dictionary[Integer, Boolean]
-                - Key = ID (CA_ID=1, SSL_ID=2, SIGNATURE_ID=3, AUTHENTICATION_ID=4)
-                - Value = used=True/unused=False
+        Constructor of the CertificateRepository 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_CERTIFICATES} ({COL_COMMON_NAME},{COL_VALID_FROM},{COL_VALID_TO},{COL_PEM_DATA},"
-               f"{COL_PRIVATE_KEY_ID},{COL_TYPE_ID},{COL_PARENT_ID})"
-               f"VALUES(?,?,?,?,?,?,?)")
-        result = DBManager.create(sql, [common_name, valid_from, valid_to, pem_data,
-                                        private_key_id, type_id, parent_id], usages)
-        return result
+        self.connection = connection
+        self.cursor = cursor
 
-    def read(self, certificate_id: int = None):
+    def create(self, certificate: Certificate) -> bool:
         """
-        Reads (selects) a certificate or certificates
+        Creates a certificate.
+        For root certificate (CA) the parent certificate id is modified to the same id (id == parent_id).
 
-        :param certificate_id: ID of specific certificate
+        :param certificate: Instance of the Certificate object
 
-        :return: instance of Certificate 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_CERTIFICATES} "
+                   f"({COL_COMMON_NAME},"
+                   f"{COL_VALID_FROM},"
+                   f"{COL_VALID_TO},"
+                   f"{COL_PEM_DATA},"
+                   f"{COL_PRIVATE_KEY_ID},"
+                   f"{COL_TYPE_ID},"
+                   f"{COL_PARENT_ID})"
+                   f"VALUES(?,?,?,?,?,?,?)")
+            values = [certificate.common_name,
+                      certificate.valid_from,
+                      certificate.valid_to,
+                      certificate.pem_data,
+                      certificate.private_key_id,
+                      certificate.type_id,
+                      certificate.parent_id]
+            self.cursor.execute(sql, values)
+            self.connection.commit()
+
+            last_id: int = self.cursor.lastrowid
+
+            if certificate.usages[ROOT_CA_ID - 1]:
+                certificate.parent_id = last_id
+                return self.update(last_id, certificate)
+            else:
+                for usage_id, usage_value in certificate.usages:
+                    if usage_value:
+                        sql = (f"INSERT INTO {TAB_CERTIFICATE_USAGES} "
+                               f"({COL_CERTIFICATE_ID},"
+                               f"{COL_USAGE_TYPE_ID}) "
+                               f"VALUES (?,?)")
+                        values = [last_id, usage_id]
+                        self.cursor.execute(sql, values)
+                        self.connection.commit()
+        except Error as e:
+            print(e)
+            return False
+
+        return True
+
+    def read(self, certificate_id: int) -> Certificate:
+        """
+        Reads (selects) a certificate.
 
-        if certificate_id is not None:
-            sql_extend = f" WHERE {COL_ID} = ?"
+        :param certificate_id: ID of specific certificate
 
-        sql = f"SELECT * FROM {TAB_CERTIFICATES}{sql_extend}"
-        certificate_rows = DBManager.read(sql, certificate_id)
+        :return: instance of the Certificate object
+        """
 
-        certificates: list = []
+        try:
+            sql = (f"SELECT * FROM {TAB_CERTIFICATES} "
+                   f"WHERE {COL_ID} = ?")
+            values = [certificate_id]
+            self.cursor.execute(sql, values)
+            certificate_row = self.cursor.fetchall()
 
-        for certificate_row in certificate_rows:
-            sql = f"SELECT * FROM {TAB_CERTIFICATE_USAGES} WHERE {COL_CERTIFICATE_ID} = ?"
-            usage_rows = DBManager.read(sql, certificate_row[0])
+            sql = (f"SELECT * FROM {TAB_CERTIFICATE_USAGES} "
+                   f"WHERE {COL_CERTIFICATE_ID} = ?")
+            self.cursor.execute(sql, values)
+            usage_rows = self.cursor.fetchall()
 
             usage_dict: Dict[int, bool] = {}
             for usage_row in usage_rows:
-                usage_dict[usage_row[1]] = True
-
-            certificates.append(Certificate(certificate_row[0],
-                                            certificate_row[1],
-                                            certificate_row[2],
-                                            certificate_row[3],
-                                            certificate_row[4],
-                                            certificate_row[5],
-                                            certificate_row[6],
-                                            certificate_row[7],
-                                            usage_dict))
-
-        if certificate_id is not None:
-            return certificates[0]
-
-        return certificates
-
-    def update(self, certificate_id: int, common_name: str = None, valid_from: str = None, valid_to: str = None,
-               pem_data: str = None, private_key_id: int = None, type_id: int = None, parent_id: int = None,
-               usages: Dict[int, bool] = None) -> bool:
+                usage_dict[usage_row[2]] = True
+
+            certificate: Certificate = Certificate(certificate_row[0],
+                                                   certificate_row[1],
+                                                   certificate_row[2],
+                                                   certificate_row[3],
+                                                   certificate_row[4],
+                                                   certificate_row[5],
+                                                   certificate_row[6],
+                                                   certificate_row[7],
+                                                   usage_dict)
+        except Error as e:
+            print(e)
+            return None
+
+        return certificate
+
+    def update(self, certificate_id: int, certificate: Certificate) -> bool:
         """
-        Updates a certificate
+        Updates a certificate.
+        If the parameter of certificate (Certificate object) is not to be changed,
+        the same value must be specified.
 
         :param certificate_id: ID of specific certificate
-        :param common_name: private key of the certificate
-        :param valid_from: string of date (time) of validation after (from)
-        :param valid_to: string of date (time) of validation before (to)
-        :param pem_data: data of certificate in a PEM format
-        :param private_key_id: ID from PrivateKeys table for specific private key
-        :param type_id: ID from CertificateTypes table (ROOT_CA_ID=1, INTERMEDIATE_CA_ID=2, CERTIFICATE_ID=3)
-        :param parent_id: ID from Certificates table for specific parent certificate
-            for root CA use e.g. -1 (ID will be changed after created certificate)
-        :param usages: dictionary of usages IDs which certificate use
-            Dictionary[Integer, Boolean]
-                - Key = ID (CA_ID=1, SSL_ID=2, SIGNATURE_ID=3, AUTHENTICATION_ID=4)
-                - Value = used=True/unused=False
+        :param certificate: Instance of the Certificate object
 
         :return: the result of whether the updation was successful
         """
 
-        updated_list = []
-        values = []
-        if common_name is not None:
-            updated_list.append(f"{COL_COMMON_NAME} = ?")
-            values.append(common_name)
-        if valid_from is not None:
-            updated_list.append(f"{COL_VALID_FROM} = ?")
-            values.append(valid_from)
-        if valid_to is not None:
-            updated_list.append(f"{COL_VALID_TO} = ?")
-            values.append(valid_to)
-        if pem_data is not None:
-            updated_list.append(f"{COL_PEM_DATA} = ?")
-            values.append(pem_data)
-        if private_key_id is not None:
-            updated_list.append(f"{COL_PRIVATE_KEY_ID} = ?")
-            values.append(private_key_id)
-        if type_id is not None:
-            updated_list.append(f"{COL_TYPE_ID} = ?")
-            values.append(type_id)
-        if parent_id is not None:
-            updated_list.append(f"{COL_PARENT_ID} = ?")
-            values.append(parent_id)
-
-        updated_str = ", ".join(updated_list)
-        sql = f"UPDATE {TAB_CERTIFICATES} SET {updated_str} WHERE {COL_ID} = ?"
-        result = DBManager.update(sql, certificate_id, values, usages)
-        return result
+        try:
+            sql = (f"UPDATE {TAB_CERTIFICATES} "
+                   f"SET {COL_COMMON_NAME} = ?, "
+                   f"{COL_VALID_FROM} = ?, "
+                   f"{COL_VALID_TO} = ?, "
+                   f"{COL_PEM_DATA} = ?, "
+                   f"{COL_PRIVATE_KEY_ID} = ?, "
+                   f"{COL_TYPE_ID} = ?, "
+                   f"{COL_PARENT_ID} = ? "
+                   f"WHERE {COL_ID} = ?")
+            values = [certificate.common_name,
+                      certificate.valid_from,
+                      certificate.valid_to,
+                      certificate.pem_data,
+                      certificate.private_key_id,
+                      certificate.type_id,
+                      certificate.parent_id]
+            self.cursor.execute(sql, values)
+            self.connection.commit()
+
+            sql = (f"DELETE FROM {TAB_CERTIFICATE_USAGES} "
+                   f"WHERE {COL_CERTIFICATE_ID} = ?")
+            values = [certificate_id]
+            self.cursor.execute(sql, values)
+            self.connection.commit()
+
+            for usage_id, usage_value in certificate.usages:
+                if usage_value:
+                    sql = (f"INSERT INTO {TAB_CERTIFICATE_USAGES} "
+                           f"({COL_CERTIFICATE_ID},"
+                           f"{COL_USAGE_TYPE_ID}) "
+                           f"VALUES (?,?)")
+                    values = [certificate_id, usage_id]
+                    self.cursor.execute(sql, values)
+                    self.connection.commit()
+        except Error as e:
+            print(e)
+            return False
+
+        return True
 
     def delete(self, certificate_id: int) -> bool:
         """
@@ -139 +171 @@
         :return: the result of whether the deletion was successful
         """
 
-        sql = f"DELETE FROM {TAB_CERTIFICATES} WHERE {COL_ID} = ?"
-        result = DBManager.delete(sql, certificate_id, True)
-        return result
+        try:
+            sql = (f"DELETE FROM {TAB_CERTIFICATES} "
+                   f"WHERE {COL_ID} = ?")
+            values = [certificate_id]
+            self.cursor.execute(sql, values)
+            self.connection.commit()
+        except Error as e:
+            print(e)
+            return False
+
+        return True