WMGZON/controllers/database/database.py

137 lines
3.8 KiB
Python

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":
__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():
setattr(obj, attr, 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