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

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:
        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):
        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:
        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:
        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