diff --git a/src/config.py b/src/config.py index 975c89e..851119d 100644 --- a/src/config.py +++ b/src/config.py @@ -10,6 +10,7 @@ def __init__(self): self.CONFIG_FILE_DIRECTORY = "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 = "" diff --git a/src/config_files/default_data.yaml b/src/config_files/default_data.yaml new file mode 100644 index 0000000..fb9cac2 --- /dev/null +++ b/src/config_files/default_data.yaml @@ -0,0 +1,4 @@ +category: + - name: 'internal_master_category' + - name: 'tesing more categories' + budget_value: '100' diff --git a/src/currency_handler.py b/src/currency_handler.py new file mode 100644 index 0000000..cb92036 --- /dev/null +++ b/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 + + +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) + diff --git a/src/data_handler.py b/src/data_handler.py index f3d9677..c1d970d 100644 --- a/src/data_handler.py +++ b/src/data_handler.py @@ -14,10 +14,13 @@ 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(self, sql_statement: str, return_id: bool = False) -> (bool, int): + def execute_sql( + 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" - logging.debug(f"received request to execute {sql_statement}") + logger.debug(f"received request to execute {sql_statement} with params {params}") new_id: int = 0 is_success: bool = False try: @@ -25,44 +28,50 @@ def execute_sql(self, sql_statement: str, return_id: bool = False) -> (bool, int database=self.database, timeout=10, # seconds i hope ) - logging.debug("opened connection to database") + logger.info("opened connection to database") cursor = connection.cursor() - cursor.execute(sql_statement, {}) + cursor.execute("PRAGMA foreign_keys = ON;") + if seed: + cursor.execute("PRAGMA foreign_keys = OFF;") + cursor.execute(sql_statement, params) connection.commit() + logger.debug(f"executed {sql_statement} with params: {params}") + logger.info("executed sql command") if return_id: new_id = cursor.lastrowid + logger.debug(f"New id created {new_id}") is_success = True except Exception as e: # TODO: better Exception handling - logging.error(e) + logger.error(e) is_success = False finally: + cursor.execute("PRAGMA foreign_keys = ON;") connection.close() + logger.info("closed connection to database") if return_id: return (is_success, new_id) else: return is_success - def read_database(self, sql_statement: str, params=()) -> list: + def read_database(self, sql_statement: str, params: tuple = ()) -> list: "reads the database query and returns the results" - logging.debug(f"received request to read {sql_statement}") + logger.debug(f"received request to read {sql_statement} with params {params}") results = [] try: connection = sqlite3.connect(self.database) cursor = connection.cursor() results = cursor.execute(sql_statement, params) results = results.fetchall() - logging.debug(f"query returned: {results}") + logger.debug(f"query returned: {results}") except Exception as e: - logging.error(f"Read error occurred: {e}") + logger.error(f"Read error occurred: {e}") finally: connection.close() return results @staticmethod def _build_column(column: dict) -> str: - logging.debug( - f"building column {column}" - ) + logging.debug(f"building column {column}") name = column["column_name"] is_pk = column["is_pk"] is_fk = column.get("is_fk") @@ -77,7 +86,6 @@ def _build_column(column: dict) -> str: column_string += " NOT NULL" return column_string - def _add_table_to_db(self, table: dict) -> bool: "Creates the table in the verity database, based on the schema yaml" sql = f"""CREATE TABLE IF NOT EXISTS {table["table_name"]} ( @@ -132,9 +140,7 @@ def print_table_schema(self, table_name): # Print the schema logger.debug(f"Schema for table: {table_name}") for row in cursor.fetchall(): - logger.debug( - f"Column Name: {row[1]}, Data Type: {row[2]}, Not Null: {row[3]}" - ) + logger.debug(f"Column Name: {row[1]}, Data Type: {row[2]}, Not Null: {row[3]}") except Exception as e: logger.error(f"An error occurred: {e}") @@ -148,79 +154,50 @@ def print_table_schema(self, table_name): 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}") - try: - connection = sqlite3.connect(self.database) - cursor = connection.cursor() - cursor.execute( - """ - INSERT INTO user ( - name - ) - VALUES (?) - """, - (user_name,), - ) - connection.commit() - user_id = cursor.lastrowid - if user_id is None: - user_id = 0 - except Exception as e: - logger.error(f"Failed to insert user name, error: {e}") - user_id = 0 - finally: - try: - connection.close() - logger.debug("connection to db closed") - return user_id - except Exception as e: - logger.error(f"failed to close connection message: {e}") - return 0 + 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) -> int: + 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}") - try: - connection = sqlite3.connect(self.database) - cursor = connection.cursor() - cursor.execute( - """ - INSERT INTO category (user_id, name, budget_value, parent_id) - VALUES (?, ?, ?, ?) - """, - (user_id, category_name, budget_value, parent_id), - ) - connection.commit() - category_id = cursor.lastrowid or 0 - except Exception as e: - logger.error(f"Failed to insert category name, error: {e}") - category_id = 0 - finally: - try: - connection.close() - except Exception as e: - logger.error(f"failed to close connection: {e}") - return category_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.""" - try: - connection = sqlite3.connect(self.database) - cursor = connection.cursor() - sql = "SELECT id, name, budget_value, parent_id FROM category WHERE user_id = ?" - cursor.execute(sql, (user_id,)) - categories = cursor.fetchall() - except Exception as e: - logger.error(f"Failed to fetch categories: {e}") - categories = [] - finally: - try: - connection.close() - except Exception as e: - logger.error(f"failed to close connection: {e}") - return categories + 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")) diff --git a/src/front/home.py b/src/front/home.py index 9d75d1f..0bf3eb0 100644 --- a/src/front/home.py +++ b/src/front/home.py @@ -2,6 +2,7 @@ from flask import Blueprint, flash, redirect, render_template, request, session, url_for +import currency_handler import data_handler from config import VerityConfig @@ -22,6 +23,7 @@ def home_page(): # Debug log to see what's being passed to the template logger.debug(f"User ID from session: {user_id}, User Name from session: {selected_user_name}") + logger.info(f"{selected_user_name} is logged in") categories = db_call.get_categories(user_id) if user_id else [] return render_template( @@ -29,33 +31,36 @@ def home_page(): users=users, categories=categories, selected_user_id=user_id, - selected_user_name=selected_user_name + selected_user_name=selected_user_name, ) + @home_bp.route("/submit_user", methods=["POST"]) def submit_user_name(): user_name = request.form.get("userName") logger.debug(f"Request Received: {request.form}") if not user_name: - flash("Please enter a user name.", "error") + flash("Please enter a user name.", "danger") return redirect(url_for("home.home_page")) db_call = data_handler.database(verity_config) # More efficient check for existing user exists = db_call.read_database("SELECT 1 FROM user WHERE name = ?", (user_name,)) if exists: - flash("A user with that name already exists.", "error") + logger.warning(f"username already exists in the database: {user_name}") + flash("A user with that name already exists.", "danger") return redirect(url_for("home.home_page")) - logger.info(f"User submitted new user name: {user_name}") user_id = db_call.add_user_name(user_name) if user_id == 0: - flash("User Name not saved, please check the logs", "error") + flash("User Name not saved, please check the logs", "danger") return redirect(url_for("home.home_page")) - - flash("user name saved! Now select the user from the dropdown to continue.", "success") + session["user_id"] = int(user_id) + session["user_name"] = user_name + flash("user name saved! Start adding categories.", "success") return redirect(url_for("home.home_page")) + @home_bp.route("/select_user", methods=["POST"]) def select_user(): selected_user_id = request.form.get("selectedUserId") @@ -85,50 +90,46 @@ def select_user(): else: session.pop("user_id", None) session.pop("user_name", None) - flash("User not found!", "error") + flash("User not found!", "danger") return redirect(url_for("home.home_page")) + @home_bp.route("/submit_category", methods=["POST"]) def submit_category(): user_id = session.get("user_id") if not user_id: - flash("No user selected!", "error") + logger.warning("No user selected and trying to add a category... how?") + flash("No user selected!", "danger") return redirect(url_for("home.home_page")) # Get and validate category name (required) category_name = request.form.get("categoryName") - if not category_name: - flash("Category name is required", "error") - return redirect(url_for("home.home_page")) - - # Handle form values more concisely - # Convert budget_value to float if provided - budget_value = None - budget_value_str = request.form.get("budgetValue", "").strip() - if budget_value_str: - try: - budget_value = float(budget_value_str) - except ValueError: - logger.error(f"Invalid budget value: {budget_value_str}") - flash("Invalid budget value - please enter a valid number", "error") - return redirect(url_for("home.home_page")) + # Convert budget_value to our universal currency + budget_value: int = 0 + budget_value_input = request.form.get("budgetValue", "0") + logger.debug(f"budget value input: {budget_value_input}") + if not budget_value_input: + logger.info("no budget value assigned to category") + else: + logger.info(f"Category:{category_name}, Assigned amount: {budget_value_input}") + # User might either add a whole currency or a decimal of. we need to handle both + if float(budget_value_input) < 0: + logger.info("user is stupid and tried to assign a negative amount to the category") + flash("Negative amounts don`t really make sense here, removed budget amount", "danger") + budget_value_input = 0 + budget_value = currency_handler.convert_to_universal_currency(budget_value_input) # Convert parent_id to int if provided - parent_id = None - parent_id_str = request.form.get("parentId", "").strip() - if parent_id_str: - try: - parent_id = int(parent_id_str) - except ValueError: - logger.error(f"Invalid parent ID: {parent_id_str}") - flash("Invalid parent category ID", "error") - return redirect(url_for("home.home_page")) + parent_id = request.form.get("parentId", "0").strip() db_call = data_handler.database(verity_config) - category_id = db_call.add_category(user_id, category_name, budget_value, parent_id) + if parent_id == 0: + category_id = db_call.add_category(user_id, category_name, budget_value) + else: + category_id = db_call.add_category(user_id, category_name, budget_value, parent_id) if category_id == 0: - flash("Category not saved, please check the logs", "error") + flash("Category not saved, please check the logs", "danger") else: flash("Category saved!", "success") return redirect(url_for("home.home_page")) diff --git a/src/tests/test_currency_handler.py b/src/tests/test_currency_handler.py new file mode 100644 index 0000000..8111018 --- /dev/null +++ b/src/tests/test_currency_handler.py @@ -0,0 +1,21 @@ + +from src import currency_handler + + +def test_convert_to_universal_currency(): + input = 87.82 + result = currency_handler.convert_to_universal_currency(input) + assert result == 8782 + + +def test_whole_number(): + input = 123 + result = currency_handler.convert_to_universal_currency(input) + assert result == input + + +def test_negative_number(): + input = -123.45 + result = currency_handler.convert_to_universal_currency(input) + print(result) + assert result == -12345 diff --git a/src/tests/test_data_handler.py b/src/tests/test_data_handler.py index becf155..283ae80 100644 --- a/src/tests/test_data_handler.py +++ b/src/tests/test_data_handler.py @@ -83,10 +83,10 @@ def test_read_database_with_params(test_db_call): def test_build_database(test_db_call): """Test that the database builds correctly with all required tables""" # Check if all tables have been created - tables_to_check = ["user", "account", "account_type", "party", "transaction_type", - "category", "transaction_log"] - for table in tables_to_check: - sql = f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table}';" + config = VerityConfig() + for table in config.DATABASE_SCHEMA["tables"]: + table_name = table["table_name"] + sql = f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table_name}';" results = test_db_call.read_database(sql) assert len(results) == 1, f"Table '{table}' should exist" @@ -182,4 +182,5 @@ def test_add_category_invalid_user(test_db_call): # Try to add a category with a non-existent user_id # (should still succeed in SQLite unless foreign keys are enforced) category_id = test_db_call.add_category(99999, "InvalidUserCategory") + assert category_id == 0 assert isinstance(category_id, int) diff --git a/src/verity.py b/src/verity.py index 69df55b..0bf866c 100644 --- a/src/verity.py +++ b/src/verity.py @@ -28,6 +28,7 @@ def set_up_logging(config): # database initialise verity = database(verity_config) + logger.debug(verity_config.DEFAULT_DATA) verity.build_database() # app initialise