Revize 1636aefe

Re #8471 - Edited files for DB communication

src/impl/certificate_repository_impl.py

@@ +1 @@
+from typing import List, Dict
+
 from ..db_objects.certificate import Certificate
 from ..dao.repository import IRepository
 from db_manager import DBManager
@@ -7 +9 @@
 class CertificateRepositoryImpl(IRepository):
 
     def create(self, common_name: str, valid_from: str, valid_to: str, pem_data: str,
-               type_id: int, private_key_id: int, usage_id: int) -> bool:
-        sql = (f"INSERT INTO {TAB_CERTIFICATE} ({COL_COMMON_NAME},{COL_VALID_FROM},{COL_VALID_TO},{COL_PEM_DATA},"
-               f"{COL_TYPE_ID},{COL_PRIVATE_KEY_ID},{COL_USAGE_ID})"
+               private_key_id: int, type_id: int, usages: Dict[int, bool]) -> bool:
+        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})"
                f"VALUES(?,?,?,?,?,?,?)")
-        result = DBManager.create(sql, common_name, valid_from, valid_to, pem_data, type_id, private_key_id, usage_id)
+        result = DBManager.create(sql, [common_name, valid_from, valid_to, pem_data, private_key_id, type_id], usages)
         return result
 
     def read(self, certificate_id: int = None):
@@ -20 +22 @@
         if certificate_id is not None:
             sql_extend = f" WHERE {COL_ID} = ?"
 
-        sql = f"SELECT * FROM {TAB_CERTIFICATE}{sql_extend}"
-        rows = DBManager.read(sql, certificate_id)
+        sql = f"SELECT * FROM {TAB_CERTIFICATES}{sql_extend}"
+        certificate_rows = DBManager.read(sql, certificate_id)
 
         certificates: list = []
 
-        for row in rows:
-            certificates.append(Certificate(row[0], row[1], row[2], row[3], row[4], row[5], row[6], row[7]))
+        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])
+
+            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],
+                                            usage_dict))
 
         if certificate_id is not None:
             return certificates[0]
@@ -34 +50 @@
         return certificates
 
     def update(self, certificate_id: int, common_name: str = None, valid_from: str = None, valid_to: str = None,
-               pem_data: str = None, type_id: int = None, private_key_id: int = None, usage_id: int = None) -> bool:
+               pem_data: str = None, private_key_id: int = None, type_id: int = None,
+               usages: Dict[int, bool] = None) -> bool:
         updated_list = []
         values = []
         if common_name is not None:
@@ -49 +66 @@
         if pem_data is not None:
             updated_list.append(f"{COL_PEM_DATA} = ?")
             values.append(pem_data)
-        if type_id is not None:
-            updated_list.append(f"{COL_TYPE_ID} = ?")
-            values.append(type_id)
         if private_key_id is not None:
             updated_list.append(f"{COL_PRIVATE_KEY_ID} = ?")
             values.append(private_key_id)
-        if usage_id is not None:
-            updated_list.append(f"{COL_USAGE_ID} = ?")
-            values.append(usage_id)
-
-        values.append(certificate_id)
+        if type_id is not None:
+            updated_list.append(f"{COL_TYPE_ID} = ?")
+            values.append(type_id)
 
         updated_str = ", ".join(updated_list)
-        sql = f"UPDATE {TAB_CERTIFICATE} SET {updated_str} WHERE {COL_ID} = ?"
-        result = DBManager.update(sql, values)
+        sql = f"UPDATE {TAB_CERTIFICATES} SET {updated_str} WHERE {COL_ID} = ?"
+        result = DBManager.update(sql, certificate_id, values, usages)
         return result
 
     def delete(self, certificate_id: int) -> bool:
-        sql = f"DELETE FROM {TAB_CERTIFICATE} WHERE {COL_ID} = ?"
-        result = DBManager.delete(sql, certificate_id)
+        sql = f"DELETE FROM {TAB_CERTIFICATES} WHERE {COL_ID} = ?"
+        result = DBManager.delete(sql, certificate_id, True)
         return result

src/impl/db_manager.py

@@ -1 +1 @@
 import sqlite3
 from sqlite3 import Error
+from typing import List, Dict
 
-from ..constants import DATABASE_FILE
+from ..constants import *
 from ..dao.repository import IRepository
 
 
 class DBManager(IRepository):
 
     @staticmethod
-    def create(sql: str, *args) -> bool:
+    def create(sql: str, values: List, usages: Dict[int, bool] = None) -> bool:
         conn = None
         try:
-            sql: str = args[0]
-            values = args[1:]
-
             conn = sqlite3.connect(DATABASE_FILE)
             c = conn.cursor()
             c.execute(sql, values)
+
+            if usages is not None:
+                for usage_id, usage_value in usages:
+                    if usage_value:
+                        c.execute(f"INSERT INTO {TAB_CERTIFICATE_USAGES} ({COL_CERTIFICATE_ID},{COL_USAGE_TYPE_ID}) "
+                                  f"VALUES (?,?)",
+                                  [c.lastrowid, usage_id])
+
             conn.commit()
             c.close()
         except Error as e:
@@ -35 +41 @@
             c = conn.cursor()
 
             if item_id is not None:
-                list_item_id = [item_id]
-                c.execute(sql, list_item_id)
+                c.execute(sql, [item_id])
             else:
                 c.execute(sql)
 
@@ -51 +56 @@
         return records
 
     @staticmethod
-    def update(sql: str, *args) -> bool:
+    def update(sql: str, item_id: int, values: List, usages: Dict[int, bool] = None) -> bool:
         conn = None
         try:
+            values.append(item_id)
+
             conn = sqlite3.connect(DATABASE_FILE)
             c = conn.cursor()
-            c.execute(sql, args)
+            c.execute(sql, values)
+
+            if usages is not None:
+                for usage_id, usage_value in usages:
+                    c.execute(f"DELETE FROM {TAB_CERTIFICATE_USAGES} "
+                              f"WHERE {COL_CERTIFICATE_ID} = ? AND {COL_USAGE_TYPE_ID} = ?",
+                              [item_id, usage_id])
+
+                    if usage_value:
+                        c.execute(f"INSERT INTO {TAB_CERTIFICATE_USAGES} ({COL_CERTIFICATE_ID},{COL_USAGE_TYPE_ID}) "
+                                  f"VALUES (?,?)",
+                                  [item_id, usage_id])
+
             conn.commit()
             c.close()
         except Error as e:
@@ -68 +87 @@
         return True
 
     @staticmethod
-    def delete(sql: str, item_id: int) -> bool:
+    def delete(sql: str, item_id: int, delete_usages: bool) -> bool:
         conn = None
         try:
             conn = sqlite3.connect(DATABASE_FILE)
             c = conn.cursor()
-            c.execute(sql, (item_id,))
+
+            if delete_usages:
+                c.execute(f"DELETE FROM {TAB_CERTIFICATE_USAGES} WHERE {COL_CERTIFICATE_ID} = ?", [item_id])
+
+            c.execute(sql, [item_id])
             conn.commit()
             c.close()
         except Error as e:

src/impl/private_key_repository_impl.py

@@ -7 +7 @@
 class PrivateKeyRepositoryImpl(IRepository):
 
     def create(self, private_key: str, password: str) -> bool:
-        sql = f"INSERT INTO {TAB_PRIVATE_KEY} ({COL_PRIVATE_KEY},{COL_PASSWORD}) VALUES(?,?)"
-        result = DBManager.create(sql, private_key, password)
+        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):
@@ -17 +17 @@
         if private_key_id is not None:
             sql_extend = f" WHERE {COL_ID} = ?"
 
-        sql = f"SELECT * FROM {TAB_PRIVATE_KEY}{sql_extend}"
+        sql = f"SELECT * FROM {TAB_PRIVATE_KEYS}{sql_extend}"
         rows = DBManager.read(sql, private_key_id)
 
         private_keys: list = []
@@ -40 +40 @@
             updated_list.append(f"{COL_PASSWORD} = ?")
             values.append(password)
 
-        values.append(private_key_id)
-
         updated_str = ", ".join(updated_list)
-        sql = f"UPDATE {TAB_PRIVATE_KEY} SET {updated_str} WHERE {COL_ID} = ?"
-        result = DBManager.update(sql, values)
+        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:
-        sql = f"DELETE FROM {TAB_PRIVATE_KEY} WHERE {COL_ID} = ?"
-        result = DBManager.delete(sql, private_key_id)
+        sql = f"DELETE FROM {TAB_PRIVATE_KEYS} WHERE {COL_ID} = ?"
+        result = DBManager.delete(sql, private_key_id, False)
         return result