aswi2021jmsd-gitlab4/src/impl/certificate_repository_impl.py @ 86415c34

from ..db_objects.certificate import Certificate
from ..dao.repository import IRepository
from db_manager import DBManager
from ..constants import *


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})"
               f"VALUES(?,?,?,?,?,?,?)")
        result = DBManager.create(sql, common_name, valid_from, valid_to, pem_data, type_id, private_key_id, usage_id)
        return result

    def read(self, certificate_id: int = None):
        sql_extend = ""

        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)

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

        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, type_id: int = None, private_key_id: int = None, usage_id: int = None) -> bool:
        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 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)

        updated_str = ", ".join(updated_list)
        sql = f"UPDATE {TAB_CERTIFICATE} SET {updated_str} WHERE {COL_ID} = ?"
        result = DBManager.update(sql, values)
        return result

    def delete(self, certificate_id: int) -> bool:
        sql = f"DELETE FROM {TAB_CERTIFICATE} WHERE {COL_ID} = ?"
        result = DBManager.delete(sql, certificate_id)
        return result