swag-shop/app/services/cart_service.py

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