Refactor to better use of classes

.github/workflows/stage_unit_tests.yml

@@ -2,7 +2,7 @@ name: Unit Tests - Stage Branch
 
 on:
   pull_request:
-    types: [opened , reopene, edited]
+    types: [opened , reopened, edited]
     branches:
       - 'stage'
 
@@ -32,5 +32,5 @@ jobs:
         run: uv run ruff check
 
       - name: Run Unit Tests
-        working-directory: ./src
+        working-directory: .
         run: uv run pytest

src/config_files/logging_config.yaml → api/config_files/logging_config.yaml

@@ -18,7 +18,7 @@ handlers:
     class: logging.handlers.RotatingFileHandler
     level: DEBUG
     formatter: withdate
-    filename: "../logs/verity.log"
+    filename: "./logs/verity.log"
     maxBytes: 10485760 # 10MB
     backupCount: 5
   queue_handler:

api/src/category.py

@@ -0,0 +1,112 @@
+import logging
+from typing import Optional
+
+from api.src.data_handler import Database
+
+# from api.src.user import User
+
+logger = logging.getLogger(__name__)
+
+
+class Category:
+    "Main category class, for anything related to the category"
+
+    def __init__(
+        self,
+        database: Database,
+        user_id: int,
+        category_name: str = "",
+        budget_value: int = 0,
+        id: int = 0,
+        parent: Optional["Category"] = None,
+    ):
+        self.database: Database = database
+        self.name: str = category_name
+        self.budget_value: int = budget_value
+        self.id: int = id
+        self.children: list[Category] = []
+        self.parent: Category = parent
+        self.user_id = user_id
+        if not self.name:
+            self.get_name()
+        logger.info(f"{self.name} / {self.id} initialised")
+
+    def __repr__(self):
+        return f"""Category:(
+        Name: {self.name}
+        Id: {self.id}
+        Budget Value: {self.budget_value}
+        Number of Childen: {len(self.children)}
+        Parent:{self.parent.name if self.parent else "Top Level Category"}
+        )"""
+
+    def __str__(self):
+        return f"Category: {self.name}"
+
+    def add(self):
+        logger.info(f"Adding {self.name} to user id {self.user_id}")
+        sql_statement = """
+        INSERT INTO category (user_id, name, budget_value, parent_id)
+        VALUES (?, ?, ?, ?)
+        """
+        if not self.parent:
+            logger.debug("No parent linked to category, linking now.")
+            self.get_default_category()
+        params = (self.user_id, self.name, self.budget_value, int(self.parent.id))
+        success, self.id = self.database.execute(sql_statement, params, return_id=True)
+        if not success:
+            logger.error(f"Failed to add Category {self.name} Check the logs")
+        return self.id
+
+    def get_default_category(self):
+        logger.info("Getting Default Category")
+        self.parent = Category(self.database, self.user_id, "internal_master_category")
+        logger.info(f"master category = {self.parent}")
+        self.parent.get_id()
+        logger.info(f"parent id = {self.parent.id}")
+
+    def get_id(self):
+        logger.info(f"Getting Category id for {self.name}")
+        category_id_sql = "SELECT id FROM category WHERE user_id = ? AND name = ?"
+        category_id_params = (self.user_id, self.name)
+        category_id = self.database.read(category_id_sql, category_id_params)
+        logger.info(f"Category id is {category_id}")
+        try:
+            self.id = category_id[0][0]
+        except IndexError:
+            self.id = 0
+
+    def get_name(self):
+        logger.info(f"Getting Name for Category {self.id}")
+        cat_name_sql = "SELECT name FROM category WHERE user_id = ? and id = ?"
+        cat_name_params = (self.user_id, self.id)
+        cat_name = self.database.read(cat_name_sql, cat_name_params)
+        logger.debug(f"Category name returned {cat_name}")
+        try:
+            self.name = cat_name[0][0]
+        except IndexError:
+            self.name = "Unknown"
+
+    def get_children(self):
+        logger.info(f"Getting Child Categories for {self.name}")
+        sql = """
+        SELECT id, name, budget_value
+        FROM category
+        WHERE user_id = ?
+        AND parent_id = ?
+        """
+        params = (self.user_id, self.id)
+        child_categories = self.database.read(sql, params)
+        for child in child_categories:
+            category = Category(self.database, self.user_id, child[1], child[2], child[0])
+            self.children.append(category)
+
+    def get(self):
+        logger.info(f"Getting details for {self.name} from database")
+        sql = """SELECT user_id, name, budget_value, id, parent_id
+        FROM category
+        WHERE id = ?
+            AND user_id = ?
+        """
+        params = (self.id, self.user_id)
+        return self.database.read(sql, params)

src/config.py → api/src/config.py

@@ -6,11 +6,10 @@
 class VerityConfig:
     def __init__(self):
         self.SECRET_KEY = os.environ.get("SECRET_KEY") or "super_secret_key"
-        self.DATABASE = "Verity.db"
-        self.CONFIG_FILE_DIRECTORY = "config_files"
+        self.DATABASE = "api/data/verity.db"
+        self.CONFIG_FILE_DIRECTORY = "api/config_files"
         self.LOGGING_CONFIG = self.load_config_file("logging_config.yaml")
         self.DATABASE_SCHEMA = self.load_config_file("verity_schema.yaml")
-        self.DEFAULT_DATA = self.load_config_file("default_data.yaml")
 
     def load_config_file(self, file):
         config = ""

api/src/currency_handler.py

@@ -0,0 +1,23 @@
+import logging
+
+logger = logging.getLogger(__name__)
+
+# TODO: Make this into a proper class so we can scale currrency handling
+
+
+class CurrencyBrain:
+    def convert_to_universal_currency(input_value: float) -> int:
+        """
+        Converts the the input value to remove all decimal places and return an int.
+        This will be the starting point for our universal currency,
+        (see docs/data_dictionary).
+        for now, we will just focus on making this an int.
+        it will need change later once we have the basics done
+        """
+        logger.info(f"received {input_value} to convert to universal currency")
+        input_value = float(input_value)
+        while input_value % 1 != 0:
+            logger.debug(f"input value is not a whole number {input_value}")
+            input_value = input_value * 10
+        logger.info(f"returning {int(input_value)}")
+        return int(input_value)

src/data_handler.py → api/src/data_handler.py

@@ -4,7 +4,7 @@
 logger = logging.getLogger(__name__)
 
 
-class database:
+class Database:
     """basic Database class to start some development
     Will need a proper refactor once basic functions are in and working
     This is POC
@@ -14,9 +14,8 @@ def __init__(self, config) -> None:
         self.verity_config = config
         self.schema = self.verity_config.DATABASE_SCHEMA
         self.database = self.verity_config.DATABASE
-        self.default_data = self.verity_config.DEFAULT_DATA
 
-    def execute_sql(
+    def execute(
         self, sql_statement: str, params: tuple = (), return_id: bool = False, seed: bool = False
     ) -> (bool, int):
         "send the query here, returns true if successful, false if fail"
@@ -49,13 +48,15 @@ def execute_sql(
             connection.close()
             logger.info("closed connection to database")
             if return_id:
+                logger.debug(f"Returning tuple (success), (new id) ({is_success},{new_id})")
                 return (is_success, new_id)
             else:
+                logger.debug(f"Returning success value {is_success}")
                 return is_success
 
-    def read_database(self, sql_statement: str, params: tuple = ()) -> list:
+    def read(self, sql_statement: str, params: tuple = ()) -> list:
         "reads the database query and returns the results"
-        logger.debug(f"received request to read {sql_statement} with params {params}")
+        logger.info(f"received request to read {sql_statement} with params {params}")
         results = []
         try:
             connection = sqlite3.connect(self.database)
@@ -151,53 +152,7 @@ def print_table_schema(self, table_name):
             except Exception as e:
                 logger.error(e)
 
-    def add_user_name(self, user_name: str) -> int:
-        "takes user name string, returns user id"
-        logger.debug(f"attempting to insert values into user table {user_name}")
-        sql_statement = """
-        INSERT INTO user (name)
-        VALUES (?)
-        """
-        params = (user_name,)
-        success, user_id = self.execute_sql(sql_statement, params, True)
-        if not success:
-            logger.error("Failed to execute sql, check the logs")
-        self.add_category(user_id, "internal_master_category", seed=True)
-        return user_id
-
     def get_users(self) -> list:
         """Returns all users in the database."""
         get_user_sql = "SELECT id, name FROM user"
-        return self.read_database(get_user_sql)
-
-    def add_category(
-        self, user_id: int, category_name: str, budget_value: int = 0, parent_id=None, seed=False
-    ) -> int:
-        """Inserts a new category. Returns the category id."""
-        logger.debug(f"attempting to insert category '{category_name}' for user {user_id}")
-        sql_statement = """
-        INSERT INTO category (user_id, name, budget_value, parent_id)
-        VALUES (?, ?, ?, ?)"""
-        if not seed:
-            if not parent_id:
-                parent_id = self.read_database(
-                    "SELECT id FROM category WHERE user_id = ? AND name = ?",
-                    (user_id, "internal_master_category"),
-                )
-                try:
-                    parent_id = parent_id[0][0]
-                except IndexError:
-                    parent_id = None
-        params = (user_id, category_name, budget_value, parent_id)
-
-        success, category_id = self.execute_sql(sql_statement, params, True, seed)
-        if not success:
-            logger.error("Failed to execute sql, check the logs")
-        return category_id
-
-    def get_categories(self, user_id: int) -> list:
-        """Returns all categories for a given user."""
-        sql = (
-            "SELECT id, name, budget_value, parent_id FROM category WHERE user_id = ? and name != ?"
-        )
-        return self.read_database(sql_statement=sql, params=(user_id, "internal_master_category"))
+        return self.read(get_user_sql)

api/src/user.py

@@ -0,0 +1,100 @@
+import logging
+from typing import List
+
+from api.src.category import Category
+from api.src.data_handler import Database
+
+logger = logging.getLogger(__name__)
+
+
+class User:
+    "Main user class, for anything related to the user"
+
+    def __init__(self, database: Database, user_name: str = "", id: int = 0):
+        self.database: Database = database
+        self.name: str = user_name
+        self.id: int = id
+        self.categories: List = []
+        self.internal_category_id = 0
+        logger.info(f"{self.name} initialised.")
+
+    def __str__(self):
+        return f"User is {self.name}"
+
+    def add(self) -> int:
+        logger.info(f"adding user {self.name}")
+        sql = "INSERT INTO USER(name) VALUES (?)"
+        params = (self.name,)
+        success, self.id = self.database.execute(sql, params, return_id=True)
+        if not success:
+            logger.error(f"Failed to insert user {self.name} Please check the logs")
+            return 0
+        category_sql = """INSERT INTO category (user_id, name, budget_value, parent_id)
+            VALUES (?, ?, ?, ?)"""
+        category_params = (self.id, "internal_master_category", 0, None)
+        category_success, default_category_id = self.database.execute(
+            category_sql, category_params, return_id=True, seed=True
+        )
+        if not category_success:
+            logger.error(f"Failed to add default category for {self.name} Please check the logs")
+        # if category fails to add,
+        # we probably need to delete the user, so we dont lock the username
+        logger.info(f"master_category id = {default_category_id}")
+        self.internal_category_id = default_category_id
+        return self.id
+
+    def exists(self) -> bool:
+        logger.info(f"Checking to see if {self.name} already exists!")
+        sql = "SELECT 1 FROM user where name = ?"
+        params = (self.name,)
+        user_in_db = self.database.read(sql, params)
+        logger.info(user_in_db)
+        if user_in_db:
+            logger.warning(f"{self.name} already exists in the database")
+            return True
+        return False
+
+    def get_categories(self):
+        "Gets all top level categories for the user"
+        logger.info(f"Getting Categories for {self.name}")
+        sql = """
+        SELECT id, name, budget_value
+        FROM category
+        WHERE user_id = ?
+        AND parent_id = ?
+        """
+        if self.internal_category_id == 0:
+            self.get_internal_master_category()
+        params = (self.id, self.internal_category_id)
+        categories = self.database.read(sql, params)
+        logger.info(f"categories: {categories}")
+        for category in categories:
+            cat = Category(self.database, self.id, category[1], category[2], category[0])
+            self.categories.append(cat)
+        return self.categories
+
+    def get(self):
+        logger.info("Getting details for User")
+        sql = """
+        SELECT name
+        FROM user
+        WHERE id = ?
+        """
+        params = (self.id,)
+        name = self.database.read(sql, params)
+        logger.info(name)
+        name = name[0][0]
+        logger.info(name)
+        self.name = name
+
+    def get_internal_master_category(self):
+        sql = """
+        SELECT id FROM category WHERE name = ? and user_id = ?
+        """
+        params = ("internal_master_category", self.id)
+        master_id = self.database.read(sql, params)
+        try:
+            master_id = master_id[0][0]
+        except Exception:
+            master_id = 0
+        self.internal_category_id = master_id