[rewrite] Initial project structure with dashboard frontend and api

backend/app/db/product_db.py

@@ -0,0 +1,118 @@
+from typing import Optional
+
+from app.extensions import db_connection
+
+from app.models.product_model import Product
+
+
+def fetch_products(page: int = 0) -> Optional[list[Product]]:
+    cursor = db_connection.cursor(dictionary=True)
+
+    offset = 10 * page
+    cursor.execute(
+        "select product.id, user.displayname as seller, product.name, product.price_pc from product inner join user on user.id = product.seller_id order by product.id limit 10 offset %s",
+        (offset,),
+    )
+    results = cursor.fetchall()
+
+    if len(results) < 1:
+        return None
+
+    result_products: list[Product] = []
+
+    for row in results:
+        result_products.append(
+            Product(
+                product_id=row["id"],
+                seller_id=row["seller_id"],
+                name=row["name"],
+                price=row["price"],
+                creation_date=row["creation_date"],
+            )
+        )
+
+    return result_products
+
+def fetch_product_by_id(product_id: int) -> Optional[Product]:
+    """
+    Fetches specific product info
+
+    :param product_id: ID of product to be updated.
+    :type product_id: int
+    """
+
+    cursor = db_connection.cursor(dictionary=True)
+
+    cursor.execute("select * from product where id = %s", (product_id,))
+    result = cursor.fetchone()
+
+    if cursor.rowcount != 1:
+        return None
+
+    result_product = Product(
+            product_id=result["id"],
+            seller_id=result["seller_id"],
+            name=result["name"],
+            price=result["price"],
+            creation_date=result["creation_date"],
+        )
+
+    return result_product
+
+
+def fetch_product_extended_by_id(product_id: int) -> Optional[Product]:
+    """
+    Fetches specific product info including the seller n
+
+    :param product_id: ID of product to be updated.
+    :type product_id: int
+    """
+
+    cursor = db_connection.cursor(dictionary=True)
+
+    cursor.execute("select * from product inner join user on user.id = product.seller_id where product.id = %s", (product_id,))
+    result = cursor.fetchone()
+
+    if cursor.rowcount != 1:
+        return None
+
+    result_product = Product(
+            product_id=result["id"],
+            seller_id=result["seller_id"],
+            seller_name=result["displayname"],
+            name=result["name"],
+            price=result["price"],
+            creation_date=result["creation_date"],
+        )
+
+    return result_product
+
+def insert_product(product: Product):
+    """
+    Creates a new product listing
+
+    :param seller_id: User ID
+    :type seller_id: str
+    :param name: New product's name
+    :type name: str
+    :param price: New product's price
+    :type price: float
+    """
+
+    cursor = db_connection.cursor()
+
+    cursor.execute(
+        "insert into product(seller_id, name, price_pc) values (%s, %s, %s)",
+        (product.seller_id, product.name, round(product.price, 2)),
+    )
+    db_connection.commit()
+
+
+def delete_product(product: Product):
+    cursor = db_connection.cursor()
+
+    cursor.execute(
+        "delete from product where id = %s",
+        (product.product_id,),
+    )
+    db_connection.commit()

backend/app/db/user_db.py

@@ -0,0 +1,79 @@
+from typing import Optional
+
+from app.extensions import db_connection
+
+from app.models.user_model import User
+
+
+def fetch_by_username(username: str) -> Optional[User]:
+    cursor = db_connection.cursor(dictionary=True)
+
+    cursor.execute("select * from user where username = %s", (username,))
+
+    result = cursor.fetchone()
+
+    result_user = (
+        User(
+            user_id=result["id"],
+            username=result["username"],
+            displayname=result["displayname"],
+            email=result["email"],
+            password=result["password"],
+            role_id=result["role_id"],
+            creation_date=result["creation_date"],
+        )
+        if result
+        else None
+    )
+
+    return result_user
+
+
+def fetch_by_id(user_id: int) -> Optional[User]:
+    cursor = db_connection.cursor(dictionary=True)
+
+    cursor.execute("select * from user where id = %s", (user_id,))
+
+    result = cursor.fetchone()
+    result_user = (
+        User(
+            user_id=result["id"],
+            username=result["username"],
+            displayname=result["displayname"],
+            email=result["email"],
+            password=result["password"],
+            role_id=result["role_id"],
+            creation_date=result["creation_date"],
+        )
+        if result
+        else None
+    )
+
+    return result_user
+
+
+def insert_user(new_user: User):
+    cursor = db_connection.cursor(dictionary=True)
+
+    cursor.execute(
+        "insert into user (username, displayname, email, password) values (%s, %s, %s, %s)",
+        (new_user.username, new_user.displayname, new_user.email, new_user.password),
+    )
+    db_connection.commit()
+
+
+def delete_user(user: User):
+    cursor = db_connection.cursor(dictionary=True)
+
+    cursor.execute("delete from user where id = %s", (user.user_id,))
+    db_connection.commit()
+
+
+def update_user(user: User):
+    cursor = db_connection.cursor(dictionary=True)
+
+    cursor.execute(
+        "update user set username=%s, displayname=%s, email=%s, password=%s where id = %s",
+        (user.username, user.displayname, user.email, user.password, user.user_id),
+    )
+    db_connection.commit()