140 lines
3.9 KiB
Python
140 lines
3.9 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") == "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
|