src/datatypes.py

"""
This file contains all the code to load the business/private datatypes from the database and interact with them.
i.e. the product categories, products, reviews etc.... including utility methods to find them, render them etc...
It is all implemented as a very simple read-only Object-Relational mapping type layer to grant access to read and
use the data in the database.  DB updates are all done by offline processes and the data setup.
"""

from __future__ import annotations  # For referencing own class

import sqlite3
from typing import List, Dict

from src.common import *


class DataLoader:
    """
    Class which controls the loading of all the data in the system from the database
    """
    active_db = "02_baseline_products_dataset"
    db_dir = os.path.join(data_dir, 'sqlite')
    db_file = os.path.join(db_dir, f"{active_db}.db")
    loaded = False

    @classmethod
    def set_db_name(cls, name: str):
        """
        Programatically switch the database in use.
        """
        if name != cls.active_db:
            new_file = os.path.join(data_dir, 'sqlite', f"{name}.db")
            cls.db_file = os.path.join(DataLoader.db_dir, f"{name}.db")
            DataLoader.load_data(reload=True)
            cls.active_db = name

    @staticmethod
    def current_db() -> str:
        """
        Get the name of the current database in use
        """
        return DataLoader.active_db[:-3]

    @staticmethod
    def available_dbs() -> List[str]:
        """
        Get the names of all the available databases in the sqlite folder
        """
        return [f[:-3] for f in os.listdir(DataLoader.db_dir) if ('products' in f) and f.endswith('.db')]

    @staticmethod
    def load_data(reload=False):
        """
        Load (or reload) all the data classes across the system from the database. Fills out all the
        class variables holding the list of all of each type, and then completes also the cross-references
        between objects so they are aware of each other (e.g. putting reviews into products and linking the reviews
        to their products)
        """
        if DataLoader.loaded and not reload:
            return

        # Wipe out any prior data
        Review.all = {}
        Feature.all = {}
        Product.all = {}
        Category.all = {}

        con = sqlite3.connect(DataLoader.db_file)
        cur = con.cursor()

        categories = cur.execute('SELECT * FROM categories').fetchall()
        for c in categories:
            Category.all[c[0]] = Category(c[0], c[1])

        features = cur.execute('SELECT * FROM features').fetchall()
        for f in features:
            feat = Feature(f[0], f[1], Category.all[f[2]])
            Feature.all[f[0]] = feat
            Category.all[f[2]].features.append(feat)

        products = cur.execute('SELECT * FROM products').fetchall()
        for p in products:
            prod = Product(p[0], p[1], p[2], p[3], Category.all[p[4]])
            Product.all[p[0]] = prod
            Category.all[p[4]].products.append(prod)

        prod_feats = cur.execute('SELECT * FROM product_features').fetchall()
        for pf in prod_feats:
            Product.all[pf[1]].features.append(Feature.all[pf[2]])
            Feature.all[pf[2]].products.append(Product.all[pf[1]])

        reviews = cur.execute('SELECT * FROM reviews').fetchall()
        for r in reviews:
            rev = Review(r[0], r[2], r[3], Product.all[r[1]])
            Review.all[r[0]] = rev
            Product.all[r[1]].reviews.append(rev)

        DataLoader.loaded = True


class Category:
    """
    A category of products (e.g. "Dishwashers"). Holds also a list of Features relevant to this
    category and a list of Products in this category
    """
    all: Dict[str, Category] = {}  # All categories by ID

    @staticmethod
    def all_sorted() -> List[Category]:
        """
        Gets a list of all the catgories, sorted by name
        """
        all_cats = list(Category.all.values())
        all_cats.sort(key=lambda x: x.name)
        return all_cats

    @staticmethod
    def by_name(name: str) -> Category:
        """
        Look up a single category by name
        """
        all_cats = list(Category.all.values())
        for c in all_cats:
            if c.name == name:
                return c

    def __init__(self, id, name):
        self.id = id
        self.name = name
        self.features = []
        self.products = []

    @property
    def feature_count(self) -> int:
        return len(self.features)

    @property
    def product_count(self) -> int:
        return len(self.products)

    @property
    def singular_name(self) -> str:
        """
        Get the name of this category, stripping the trailing s if there is one
        """
        if self.name[-1] == "s":
            return self.name[:-1]  # Clip the s
        return self.name

    @property
    def lower_singular_name(self):
        """
        Get the name of this ctaegory in lower case, removing any trailing s
        """
        if self.name[-1] == "s":
            return self.name[:-1].lower()  # Clip the s
        return self.name.lower()


class Feature:
    """
    A feature available to Products in a given Category (e.g. "Quiet operation")
    """
    all: Dict[str, Feature] = {}  # All features by ID

    def __init__(self, id, name, category):
        self.id = id
        self.name = name
        self.category = category
        self.products = []

    @property
    def product_count(self) -> int:
        return len(self.products)

    def __repr__(self) -> str:
        return self.name


class Product:
    """
    A product specification with its price, description, features, and a list of associated
    customer reviews.
    """
    all: Dict[str, Product] = {}  # All products by ID

    def __init__(self, id, name, description, price, category):
        self.id = id
        self.name = name
        self.description = description
        self.price = round(price, 2)
        self.category = category
        self.features = []
        self.reviews = []

    @property
    def feature_count(self) -> int:
        return len(self.features)

    @property
    def review_count(self) -> int:
        return len(self.reviews)

    @property
    def average_rating(self, decimals=2) -> float:
        if self.review_count == 0:
            return 0.0
        return float(round(sum([r.rating for r in self.reviews]) / self.review_count, decimals))

    @staticmethod
    def for_ids(ids: List[str]) -> List[Product]:
        """
        Get a set of products based on a set of IDs
        """
        return[Product.all[i] for i in ids]

    @staticmethod
    def all_as_list() -> List[Product]:
        return list(Product.all.values())


class Review:
    """
    A single customer review with its star rating and the review text
    """
    all: Dict[str, Review] = {}

    def __init__(self, id, rating, review_text, product):
        self.id = id
        self.rating = rating
        self.review_text = review_text
        self.product = product