aswi2021jmsd-gitlab4/src/impl/db_manager.py @ a0602bad

import sqlite3
from sqlite3 import Error
from typing import List, Dict

from ..constants import *
from ..dao.repository import IRepository


class DBManager(IRepository):

    @staticmethod
    def create(sql: str, values: List, usages: Dict[int, bool] = None) -> bool:
        """
        Creates an item in database

        :param sql: SQL query for creation (parameter values are replaced by question mark)
        :param values: list of parameter values
        :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

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

        conn = None
        try:
            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:
            print(e)
            return False
        finally:
            if conn:
                conn.close()
        return True

    @staticmethod
    def read(sql: str, item_id: int = None):
        """
        Reads (selecs) an item or items from database

        :param sql: SQL query for selection (parameter values are replaced by question mark)
        :param item_id: ID of specific item

        :return: list of rows selected from database
        """

        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

    @staticmethod
    def update(sql: str, item_id: int, values: List, usages: Dict[int, bool] = None) -> bool:
        """
        Updates an item in database

        :param sql: SQL query for creation (parameter values are replaced by question mark)
        :param item_id: ID of specific item
        :param values: list of parameter values
        :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

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

        conn = None
        try:
            values.append(item_id)

            conn = sqlite3.connect(DATABASE_FILE)
            c = conn.cursor()
            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:
            print(e)
            return False
        finally:
            if conn:
                conn.close()
        return True

    @staticmethod
    def delete(sql: str, item_id: int, delete_usages: bool) -> bool:
        """
        Deletes an item from database

        :param sql: SQL query for selection (parameter values are replaced by question mark)
        :param item_id: ID of specific item
        :param delete_usages: flag specifying whether to erase rows with specific id
            from the table of usages (CertificateUsages)

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

        conn = None
        try:
            conn = sqlite3.connect(DATABASE_FILE)
            c = conn.cursor()

            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:
            print(e)
            return False
        finally:
            if conn:
                conn.close()
        return True