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
	def add_to_cart(user_id: str, product_id: int, count: int) -> Tuple[Union[dict, str], int]:
		try:
			with db_connection.cursor(dictionary=True) as cursor:
				cursor.execute("select count from cart_item where cart_id = %s and product_id = %s", (user_id, product_id))
				result = cursor.fetchone()

				if cursor.rowcount != 0:
					cursor.execute("update cart_item set count = count + %s where cart_id = %s and product_id = %s", (count, user_id, product_id))
				else:
					cursor.execute("insert into cart_item(cart_id, product_id, count) values (%s, %s, %s)", (user_id, product_id, count))

				db_connection.commit()

				return {"Success": "Successfully added to cart"}, 200

		except Error as e:
			return {"Failed": f"Failed to add item to cart. Reason: {e}"}, 400

	@staticmethod
	def update_count(user_id: str, product_id: int, count: int) -> 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}"}, 400

	@staticmethod
	def delete_from_cart(user_id: str, product_id: int) -> 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}"}, 400


	@staticmethod
	def show_cart(user_id: str) -> 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}"}, 400


	@staticmethod
	def purchase(user_id: str) -> 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 {"Failed": f"Failed to load cart. Reason: {e}"}, 400

		return {"Success": "Successfully purchased"}, 200