[rewrite] Changes services to crud, modified routes abit

backend/app/crud/cart_service.py

@@ -0,0 +1,162 @@
+from mysql.connector import Error
+from typing import Tuple, Union
+
+from app.extensions import db_connection
+
+
+class CartService:
+
+    @staticmethod
+    @staticmethod
+    def update_count(
+        user_id: str, product_id: int, count: int
+    ) -> Tuple[Union[dict, str], int]:
+        """
+        Updates count of products in user's cart
+
+        :param user_id: User ID.
+        :type user_id: str
+        :param product_id: ID of product to be updated.
+        :type product_id: int
+        :param count: New count of products
+        :type count: int
+        :return: Tuple containing a dictionary with a token and an HTTP status code.
+        :rtype: Tuple[Union[dict, str], int]
+        """
+
+        try:
+            if count <= 0:
+                return CartService.delete_from_cart(user_id, product_id)
+
+            with db_connection.cursor(dictionary=True) as cursor:
+                cursor.execute(
+                    "update cart_item set count = %s where cart_id = %s and product_id = %s",
+                    (count, user_id, product_id),
+                )
+                db_connection.commit()
+
+                return {"Success": "Successfully added to cart"}, 200
+
+        except Error as e:
+            return {"Failed": f"Failed to update item count in cart. Reason: {e}"}, 500
+
+    @staticmethod
+    def delete_from_cart(user_id: str, product_id: int) -> Tuple[Union[dict, str], int]:
+        """
+        Completely deletes an item from a user's cart
+
+        :param user_id: User ID.
+        :type user_id: str
+        :param product_id: ID of product to be updated.
+        :type product_id: int
+        :return: Tuple containing a dictionary with a token and an HTTP status code.
+        :rtype: Tuple[Union[dict, str], int]
+        """
+
+        try:
+            with db_connection.cursor() as cursor:
+                cursor.execute(
+                    "delete from cart_item where cart_id = %s and product_id = %s",
+                    (user_id, product_id),
+                )
+                db_connection.commit()
+
+                return {"Success": "Successfully removed item from cart"}, 200
+        except Error as e:
+            return {"Failed": f"Failed to remove item from cart. Reason: {e}"}, 500
+
+    @staticmethod
+    def show_cart(user_id: str) -> Tuple[Union[dict, str], int]:
+        """
+        Gives the user the content of their cart
+
+        :param user_id: User ID.
+        :type user_id: str
+        :return: Tuple containing a dictionary with a token and an HTTP status code.
+        :rtype: Tuple[Union[dict, str], int]
+        """
+
+        try:
+            with db_connection.cursor(dictionary=True) as cursor:
+                cursor.execute(
+                    "select product.name as product_name, count, price_subtotal, date_added from cart_item inner join product on cart_item.product_id = product.id where cart_item.cart_id = %s",
+                    (user_id,),
+                )
+                rows = cursor.fetchall()
+
+                results = []
+
+                for row in rows:
+                    mid_result = {
+                        "name": row["product_name"],
+                        "count": row["count"],
+                        "price_subtotal": row["price_subtotal"],
+                        "date_added": row["date_added"],
+                    }
+
+                    results.append(mid_result)
+
+            return results, 200
+
+        except Error as e:
+            return {"Failed": f"Failed to load cart. Reason: {e}"}, 500
+
+    @staticmethod
+    def purchase(user_id: str) -> Tuple[Union[dict, str], int]:
+        """
+        "Purchases" the contents of user's cart
+
+        :param user_id: User ID.
+        :type user_id: str
+        :return: Tuple containing a dictionary with a token and an HTTP status code.
+        :rtype: Tuple[Union[dict, str], int]
+        """
+
+        try:
+            with db_connection.cursor(dictionary=True) as cursor:
+                # get all cart items
+                cursor.execute(
+                    "select id, product_id, count, price_subtotal from cart_item where cart_id = %s",
+                    (user_id,),
+                )
+                results = cursor.fetchall()
+
+                if len(results) < 1:
+                    return {"Failed": "Failed to purchase. Cart is Empty"}, 400
+
+                # create a purchase
+                cursor.execute("insert into purchase(user_id) values (%s)", (user_id,))
+
+                last_id = cursor.lastrowid
+
+                parsed = []
+                ids = []
+
+                for row in results:
+                    mid_row = (
+                        last_id,
+                        row["product_id"],
+                        row["count"],
+                        row["price_subtotal"],
+                    )
+
+                    row_id = row["id"]
+
+                    parsed.append(mid_row)
+                    ids.append(row_id)
+
+                insert_query = "INSERT INTO purchase_item (purchase_id, product_id, count, price_subtotal) VALUES (%s, %s, %s, %s)"
+                for row in parsed:
+                    cursor.execute(insert_query, row)
+
+                delete_query = "delete from cart_item where id = %s"
+                for one_id in ids:
+                    cursor.execute(delete_query, (one_id,))
+
+                db_connection.commit()
+
+        # clear cart
+        except Error as e:
+            return {"msg": f"Failed to load cart. Reason: {e}"}, 500
+
+        return {"msg": "Successfully purchased"}, 200

backend/app/crud/product/product_create_service.py

@@ -0,0 +1,30 @@
+from mysql.connector import Error as mysqlError
+
+from app.messages.api_responses import product_responses as response
+import app.messages.api_errors as errors
+
+from app.db import product_db
+
+from app.models.product_model import Product
+
+
+def create_product(seller_id: str, name: str, price: float):
+    """
+    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
+    """
+
+    product: Product = Product(seller_id=seller_id, name=name, price=price)
+    try:
+        product_db.insert_product(product)
+
+    except mysqlError as e:
+        return errors.UNKNOWN_DATABASE_ERROR(e)
+
+    return response.PRODUCT_LISTING_CREATED_SUCCESSFULLY

backend/app/crud/product/product_delete_service.py

@@ -0,0 +1,23 @@
+from mysql.connector import Error as mysqlError
+
+from app.messages.api_responses import product_responses as response
+import app.messages.api_errors as errors
+
+from app.db import product_db
+
+from app.models.product_model import Product
+
+
+def delete_product(seller_id: str, product_id: str):
+    product: Product = product_db.fetch_product_by_id(product_id)
+
+    if product.seller_id != seller_id:
+        return response.NOT_OWNER_OF_PRODUCT
+
+    try:
+        product_db.delete_product(product)
+
+    except mysqlError as e:
+        return errors.UNKNOWN_DATABASE_ERROR(e)
+
+    return response.PRODUCT_LISTING_CREATED_SUCCESSFULLY

backend/app/crud/product/product_helper.py

@@ -0,0 +1,9 @@
+import imghdr
+
+
+def is_base64_jpg(decoded_string) -> bool:
+    try:
+        image_type = imghdr.what(None, decoded_string)
+        return image_type == "jpeg"
+    except Exception:
+        return False

backend/app/crud/product/product_info_service.py

@@ -0,0 +1,20 @@
+from mysql.connector import Error as mysqlError
+
+from app.messages.api_responses import product_responses as response
+import app.messages.api_errors as errors
+
+from app.db import product_db
+
+from app.models.product_model import Product
+
+
+def product_info(product_id: int):
+    try:
+        product: Product = product_db.fetch_product_extended_by_id(product_id)
+
+        if product is None:
+            return response.UNKNOWN_PRODUCT
+
+        return product, 200
+    except mysqlError as e:
+        return errors.UNKNOWN_DATABASE_ERROR(e)

backend/app/crud/product/product_list_service.py

@@ -0,0 +1,30 @@
+from mysql.connector import Error as mysqlError
+
+from app.messages.api_responses import product_responses as response
+import app.messages.api_errors as errors
+
+from app.db import product_db
+
+
+def product_list(page: int):
+    try:
+        result_products = product_db.fetch_products(page)
+
+        if result_products is None:
+            return response.SCROLLED_TOO_FAR
+
+        result_obj = []
+        for product in result_products:
+            mid_result = {
+                "id": product.product_id,
+                "seller": product.seller_id,
+                "name": product.name,
+                "price": product.price,
+            }
+
+            result_obj.append(mid_result)
+
+        return result_obj, 200
+
+    except mysqlError as e:
+        errors.UNKNOWN_DATABASE_ERROR(e)

backend/app/crud/user/delete_service.py

@@ -0,0 +1,32 @@
+from typing import Tuple, Union
+from mysql.connector import Error as mysqlError
+
+import app.db.user_db as user_db
+from app.models.user_model import User
+
+import app.messages.api_responses.user_responses as response
+import app.messages.api_errors as errors
+from app.mail.mail import send_mail
+
+
+from app.messages.mail_responses.user_email import USER_EMAIL_SUCCESSFULLY_DELETED_ACCOUNT
+
+
+def delete_user(user_id: str) -> Tuple[Union[dict, str], int]:
+    """
+    Deletes a user account.
+
+    :param user_id: User ID.
+    :type user_id: str
+    :return: Tuple containing a dictionary and an HTTP status code.
+    :rtype: Tuple[Union[dict, str], int]
+    """
+
+    try:
+        user: User = user_db.fetch_by_id(user_id=user_id)
+        user_db.delete_user(user)
+        send_mail(USER_EMAIL_SUCCESSFULLY_DELETED_ACCOUNT, user.email)
+    except mysqlError as e:
+        return errors.UNKNOWN_DATABASE_ERROR(e)
+
+    return response.USER_DELETED_SUCCESSFULLY

backend/app/crud/user/login_service.py

@@ -0,0 +1,44 @@
+import datetime
+from typing import Tuple, Union
+
+import bcrypt
+from mysql.connector import Error as mysqlError
+from flask_jwt_extended import create_access_token
+
+import app.messages.api_responses.user_responses as response
+import app.messages.api_errors as errors
+from app.db import user_db
+from app.mail.mail import send_mail
+from app.models.user_model import User
+from app.messages.mail_responses.user_email import USER_EMAIL_SUCCESSFULLY_LOGGED_IN
+
+
+def login(username: str, password: str) -> Tuple[Union[dict, str], int]:
+    """
+    Authenticates a user with the provided username and password.
+
+    :param username: User's username.
+    :type username: str
+    :param password: User's password.
+    :type password: str
+    :return: Tuple containing a dictionary with a token and an HTTP status code.
+    :rtype: Tuple[Union[dict, str], int]
+    """
+    try:
+        user: User = user_db.fetch_by_username(username)
+
+        if user is None:
+            return response.USERNAME_NOT_FOUND
+
+        if not bcrypt.checkpw(password.encode("utf-8"), user.password.encode("utf-8")):
+            return response.INCORRECT_PASSWORD
+
+        expire = datetime.timedelta(hours=1)
+        token = create_access_token(identity=user.user_id, expires_delta=expire)
+
+        send_mail(USER_EMAIL_SUCCESSFULLY_LOGGED_IN, user.email)
+
+        return {"token": token}, 200
+
+    except mysqlError as e:
+        return errors.UNKNOWN_DATABASE_ERROR(e)

backend/app/crud/user/logout_service.py

@@ -0,0 +1,31 @@
+from typing import Tuple, Union
+
+import app.messages.api_responses.user_responses as response
+from app.db import user_db
+from app.models.user_model import User
+from app.mail.mail import send_mail
+from app.services.user import user_helper as helper
+from app.messages.mail_responses.user_email import USER_EMAIL_SUCCESSFULLY_LOGGED_OUT
+
+
+def logout(jwt_token, user_id, send_notif: bool) -> Tuple[Union[dict, str], int]:
+    """
+    Logs out a user by invalidating the provided JWT.
+
+    :param jti: JWT ID.
+    :type jti: str
+    :param exp: JWT expiration timestamp.
+    :type exp: int
+    :return: Tuple containing a dictionary and an HTTP status code.
+    :rtype: Tuple[Union[dict, str], int]
+    """
+
+    jti = jwt_token["jti"]
+    exp = jwt_token["exp"]
+
+    user: User = user_db.fetch_by_id(user_id)
+
+    helper.invalidate_token(jti, exp)
+    if send_notif:
+        send_mail(USER_EMAIL_SUCCESSFULLY_LOGGED_OUT, user.email)
+    return response.USER_LOGGED_OUT_SUCCESSFULLY

backend/app/crud/user/register_service.py

@@ -0,0 +1,64 @@
+import bcrypt
+from typing import Tuple, Union
+from mysql.connector import Error as mysqlError
+
+import app.messages.api_responses.user_responses as response
+import app.messages.api_errors as errors
+from app.db import user_db
+from app.mail.mail import send_mail
+from app.models.user_model import User
+from app.services.user import user_helper as helper
+from app.messages.mail_responses.user_email import USER_EMAIL_SUCCESSFULLY_REGISTERED
+
+
+def register(
+    username: str, displayname: str, email: str, password: str
+) -> Tuple[Union[dict, str], int]:
+    """
+    Registers a new user with the provided username, email, and password.
+
+    :param username: User's username.
+    :type username: str
+    :param email: User's email address.
+    :type email: str
+    :param password: User's password.
+    :type password: str
+    :return: Tuple containing a dictionary and an HTTP status code.
+    :rtype: Tuple[Union[dict, str], int]
+    """
+
+    try:
+        if not helper.verify_username(username):
+            return response.INVALID_USERNAME_FORMAT
+
+        if not helper.verify_displayname(displayname):
+            return response.INVALID_DISPLAYNAME_FORMAT
+
+        if not helper.verify_email(email):
+            return response.INVALID_EMAIL_FORMAT
+
+        if not helper.verify_password(password):
+            return response.INVALID_PASSWORD_FORMAT
+
+        hashed_password = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt())
+
+        new_user: User = User(
+            username=username,
+            displayname=displayname,
+            email=email,
+            password=hashed_password,
+        )
+
+        user_db.insert_user(new_user)
+
+    except mysqlError as e:
+        if "email" in e.msg:
+            return response.EMAIL_ALREADY_IN_USE
+        if "username" in e.msg:
+            return response.USERNAME_ALREADY_IN_USE
+
+        return errors.UNKNOWN_DATABASE_ERROR(e)
+
+    send_mail(USER_EMAIL_SUCCESSFULLY_REGISTERED, new_user.email)
+
+    return response.USER_CREATED_SUCCESSFULLY

backend/app/crud/user/update_user_service.py

@@ -0,0 +1,72 @@
+import bcrypt
+from typing import Tuple, Union
+from mysql.connector import Error as mysqlError
+
+from app.db import user_db
+from app.mail.mail import send_mail
+from app.models.user_model import User
+from app.services.user import user_helper as helper
+from app.messages.api_responses import user_responses as response
+import app.messages.api_errors as errors
+from app.messages.mail_responses.user_email import (
+    USER_EMAIL_SUCCESSFULLY_UPDATED_ACCOUNT,
+)
+
+
+def update_user(
+    user_id: str,
+    new_username: str = None,
+    new_displayname: str = None,
+    new_email: str = None,
+    new_password: str = None,
+) -> Tuple[Union[dict, str], int]:
+    user: User = user_db.fetch_by_id(user_id)
+
+    updated_attributes = []
+
+    if user is None:
+        return response.UNKNOWN_ERROR
+
+    if new_username:
+        if not helper.verify_username(new_username):
+            return response.INVALID_USERNAME_FORMAT
+
+        user.username = new_username
+        updated_attributes.append("username")
+
+    if new_displayname:
+        if not helper.verify_displayname(new_displayname):
+            return response.INVALID_DISPLAYNAME_FORMAT
+
+        user.displayname = new_displayname
+        updated_attributes.append("displayname")
+
+    if new_email:
+        if not helper.verify_email(new_email):
+            return response.INVALID_EMAIL_FORMAT
+
+        user.email = new_email
+        updated_attributes.append("email")
+
+    if new_password:
+        if not helper.verify_password(new_password):
+            return response.INVALID_PASSWORD_FORMAT
+
+        hashed_password = bcrypt.hashpw(new_password.encode("utf-8"), bcrypt.gensalt())
+
+        user.password = hashed_password
+        updated_attributes.append("password")
+
+    try:
+        user_db.update_user(user)
+
+    except mysqlError as e:
+        if "username" in e.msg:
+            return response.USERNAME_ALREADY_IN_USE
+        if "email" in e.msg:
+            return response.EMAIL_ALREADY_IN_USE
+
+        return errors.UNKNOWN_DATABASE_ERROR(e)
+
+    send_mail(USER_EMAIL_SUCCESSFULLY_UPDATED_ACCOUNT, user.email)
+    return response.USER_ACCOUNT_UPDATED_SUCCESSFULLY(updated_attributes)

backend/app/crud/user/user_helper.py

@@ -0,0 +1,74 @@
+import re
+from datetime import datetime
+
+from app.extensions import jwt_redis_blocklist
+
+
+def invalidate_token(jti: str, exp: int):
+    """
+    Invalidates a JWT by adding its JTI to the Redis blocklist.
+
+    :param jti: JWT ID.
+    :type jti: str
+    :param exp: JWT expiration timestamp.
+    :type exp: int
+    """
+    expiration = datetime.fromtimestamp(exp)
+    now = datetime.now()
+
+    delta = expiration - now
+    jwt_redis_blocklist.set(jti, "", ex=delta)
+
+
+def verify_email(email: str) -> bool:
+    """
+    Verifies a given email string against a regular expression.
+
+    :param email: Email string.
+    :type email: str
+    :return: Boolean indicating whether the email successfully passed the check.
+    :rtype: bool
+    """
+    email_regex = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
+    return re.match(email_regex, email) and len(email) <= 64
+
+
+def verify_displayname(displayname: str) -> bool:
+    """
+    Verifies a given display name string against a regular expression.
+
+    :param displayname: Display name string.
+    :type displayname: str
+    :return: Boolean indicating whether the display name successfully passed the check.
+    :rtype: bool
+    """
+    displayname_regex = r"^[a-zA-Z.-_]{1,64}$"
+    return re.match(displayname_regex, displayname)
+
+
+def verify_username(username: str) -> bool:
+    """
+    Verifies a given username string against a regular expression.
+
+    :param username: Username string.
+    :type username: str
+    :return: Boolean indicating whether the username successfully passed the check.
+    :rtype: bool
+    """
+    username_regex = r"^[a-z]{1,64}$"
+    return re.match(username_regex, username)
+
+
+def verify_password(password: str) -> bool:
+    """
+    Verifies a given password string against a regular expression.
+
+    :param password: Password string.
+    :type password: str
+    :return: Boolean indicating whether the password successfully passed the check.
+    :rtype: bool
+    """
+    password_regex = (
+        r"^(?=.*?[A-Z])(?=.*?[a-z])(?=.*?[0-9])(?=.*?[#?!@$ %^&*-]).{8,64}$"
+    )
+    return re.match(password_regex, password)

backend/app/crud/user_crud.py

@@ -0,0 +1,23 @@
+from typing import Optional
+from sqlmodel import Session, select
+
+from app.database.models.user_model import User
+from app.core.security import verify_password
+from app.utils.models import generate_user_uuid5
+
+def get_user_by_generated_uuid(session: Session, email: str, shop_id: Optional[int]) -> Optional[User]:
+    user_uuid = generate_user_uuid5(email, shop_id)
+    stmt = select(User).where(User.uuid == user_uuid)
+    db_user = session.exec(stmt).one_or_none()
+    return db_user
+
+def create_user(session: Session):
+    raise NotImplementedError()
+
+def authenticate(session: Session, email: str, password: str, shop_id: Optional[int]) -> Optional[User]:
+    db_user = get_user_by_generated_uuid(session, email, shop_id)
+    if db_user is None:
+        return None
+    if not verify_password(plain_password=password, hashed_password=db_user.password):
+        return None
+    return db_user