WMGZON/controllers/database/database.py

from abc import ABC, abstractmethod
from typing import Mapping, Any
import sqlite3
import os


class DatabaseController(ABC):
    """ Abstract Base Class to handle database access for each component
        in the web app
    """
    __data_dir = "./data/"
    __db_name = "wmgzon.db"

    # Use test file if necessary
    if os.environ.get("ENVIRON", "test") == "test":
        __db_name = "test_" + __db_name

    __sqlitefile = __data_dir + __db_name

    def __init__(self):
        """ Initialises the object and creates a connection to the local
            DB on the server
        """
        self._conn = None
        try:
            # Creates a connection and specifies a flag to parse all types
            # back down into Python declared types e.g. date & time
            self._conn = sqlite3.connect(
                self.__sqlitefile, detect_types=sqlite3.PARSE_DECLTYPES)
        except sqlite3.Error as e:
            # Close the connection if still open
            if self._conn:
                self._conn.close()
            print(e)

    def __del__(self):
        """ Object Destructor which kills the connection to the database """
        if self._conn is not None:
            self._conn.close()

    def new_instance(self, of: type, with_fields: Mapping[str, Any]):
        """ Takes a dictionary of fields and returns the object
            with those fields populated
        """
        obj = of.__new__(of)
        for attr, value in with_fields.items():
            try:
                setattr(obj, attr, value)
            except AttributeError:
                return of(value)
        return obj

    def do(self, query: str, params: list[str]):
        """ Function to run a query that returns no response """
        self._conn.execute(
            query,
            params
        )
        self._conn.commit()

    def get_one(self, query: str, params: list[str], type: type = None
                ) -> type | None:
        """ Returns one item from the given query """
        if type is None:
            type = self.TYPE

        cursor = self._conn.execute(
            query,
            params
        )
        row = cursor.fetchone()

        if row is None:
            return None

        # Construct the row into a single object
        params = dict(zip(self.FIELDS, row))
        obj = self.new_instance(type, params)

        return obj

    def get_many(self, query: str, params: list[str], type: type = None
                 ) -> list[type] | None:
        """ Returns all items matching the given query """
        if type is None:
            type = self.TYPE

        cursor = self._conn.execute(
            query,
            params
        )
        rows = cursor.fetchall()

        if rows is None or len(rows) == 0:
            return None

        objs = list()

        # Construct the row into a list of object
        for row in rows:
            params = dict(zip(self.FIELDS, row))
            obj = self.new_instance(type, params)
            objs.append(obj)

        return objs

    """
        Set of CRUD methods to allow for Data manipulation on the backend

        These items MUST be implemented by anything inheriting from this class
    """

    @abstractmethod
    def create(self):
        """ Abstract method used to create a new record of a given
            type within the database
        """
        pass

    @abstractmethod
    def read(self):
        """ Abstract method used to read a record of a given
            type from the database
        """
        pass

    @abstractmethod
    def update(self):
        """ Abstract method used to update a record of a given
            type within the database
        """
        pass

    @abstractmethod
    def delete(self):
        """ Abstract method used to delete record of a given
            type from the database
        """
        pass