Revize 2c96b4a5

Re #8471 - Implemented DB communication
- implemented operations create, read, update, delete
- created file with constants contains name of DB file and default values in DB

src/constants.py

@@ +1 @@
+DATABASE_FILE = r"C:\sqlite\db\pythonsqlite.db"
+
+ROOT_CA_ID = 1
+INTERMEDIATE_CA_ID = 2
+CERTIFICATE_ID = 3
+
+CA_ID = 1
+SSL_ID = 2
+SIGNATURE_ID = 3
+AUTHENTICATION_ID = 4

src/impl/db_manager.py

@@ +1 @@
+import sqlite3
+from sqlite3 import Error
+
+from ..constants import DATABASE_FILE
+from ..dao.repository import IRepository
+
+
+class DBManager(IRepository):
+
+    def create(self, sql: str, *args) -> bool:
+        conn = None
+        try:
+            sql: str = args[0]
+            values = args[1:]
+
+            conn = sqlite3.connect(DATABASE_FILE)
+            c = conn.cursor()
+            c.execute(sql, values)
+            conn.commit()
+            c.close()
+        except Error as e:
+            print(e)
+            return False
+        finally:
+            if conn:
+                conn.close()
+        return True
+
+    def read(self, sql: str, item_id: int = None) -> list:
+        conn = None
+        try:
+            conn = sqlite3.connect(DATABASE_FILE)
+            c = conn.cursor()
+
+            if item_id is not None:
+                c.execute(sql, item_id)
+            else:
+                c.execute(sql)
+
+            records = c.fetchall()
+            c.close()
+        except Error as e:
+            print(e)
+            return []
+        finally:
+            if conn:
+                conn.close()
+        return records
+
+    def update(self, sql: str, item_id: int, *args) -> bool:
+        conn = None
+        try:
+            values = list(args)
+            values.append(item_id)
+
+            conn = sqlite3.connect(DATABASE_FILE)
+            c = conn.cursor()
+            c.execute(sql, values)
+            conn.commit()
+            c.close()
+        except Error as e:
+            print(e)
+            return False
+        finally:
+            if conn:
+                conn.close()
+        return True
+
+    def delete(self, sql: str, item_id: int) -> bool:
+        conn = None
+        try:
+            conn = sqlite3.connect(DATABASE_FILE)
+            c = conn.cursor()
+            c.execute(sql, (item_id,))
+            conn.commit()
+            c.close()
+        except Error as e:
+            print(e)
+            return False
+        finally:
+            if conn:
+                conn.close()
+        return True