Cart and purchase
This commit is contained in:
parent
ca8366d950
commit
e6ed37ad5e
@ -5,12 +5,55 @@ from app.api import bp_cart
|
||||
|
||||
from app.services.cart_service import CartService
|
||||
|
||||
@bp_cart.route('add/<int:product_id>', methods=['PUT'])
|
||||
@bp_cart.route('', methods=['GET'])
|
||||
@jwt_required()
|
||||
def show_cart():
|
||||
user_id = get_jwt_identity()
|
||||
|
||||
result, status_code = CartService.show_cart(user_id)
|
||||
|
||||
return result, status_code
|
||||
|
||||
@bp_cart.route('/add/<int:product_id>', methods=['PUT'])
|
||||
@jwt_required()
|
||||
def add_to_cart(product_id: int):
|
||||
user_id = get_jwt_identity()
|
||||
count = request.args.get('count', default=1, type=int)
|
||||
|
||||
if count < 1:
|
||||
return abort(400)
|
||||
|
||||
result, status_code = CartService.add_to_cart(user_id, product_id, count)
|
||||
|
||||
return result, status_code
|
||||
|
||||
@bp_cart.route('/remove/<int:product_id>', methods=['DELETE'])
|
||||
@jwt_required()
|
||||
def remove_from_cart(product_id: int):
|
||||
user_id = get_jwt_identity()
|
||||
|
||||
result, status_code = CartService.delete_from_cart(user_id, product_id)
|
||||
|
||||
return result, status_code
|
||||
|
||||
@bp_cart.route('/update/<int:product_id>', methods=['PUT'])
|
||||
@jwt_required()
|
||||
def update_count_in_cart(product_id: int):
|
||||
user_id = get_jwt_identity()
|
||||
count = request.args.get('count', type=int)
|
||||
|
||||
if not count:
|
||||
return abort(400)
|
||||
|
||||
result, status_code = CartService.update_count(user_id, product_id, count)
|
||||
|
||||
return result, status_code
|
||||
|
||||
@bp_cart.route('/purchase', methods=['GET'])
|
||||
@jwt_required()
|
||||
def purchase():
|
||||
user_id = get_jwt_identity()
|
||||
|
||||
result, status_code = CartService.purchase(user_id)
|
||||
|
||||
return result, status_code
|
||||
@ -1,30 +1,127 @@
|
||||
from mysql.connector import Error
|
||||
from typing import Tuple, Union
|
||||
|
||||
from app.extensions import db_cursor, db_connection
|
||||
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() as cursor:
|
||||
db_connection.begin()
|
||||
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()
|
||||
|
||||
db_cursor.execute("select * from cart_item where cart_id = %s and product_id = %s", (user_id, product_id))
|
||||
|
||||
if db_cursor.rowcount != 0:
|
||||
db_cursor.execute("update cart_item set count = count + %s where cart_id = %s and product_id = %s", (count, user_id, product_id))
|
||||
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:
|
||||
db_cursor.execute("insert into cart_item(cart_id, product_id, count) values (%s, %s, %s)", (user_id, product_id, count))
|
||||
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. {e}"}
|
||||
return {"Failed": f"Failed to add item to cart. Reason: {e}"}, 400
|
||||
|
||||
@staticmethod
|
||||
def show_cart(user_id: str):
|
||||
return None
|
||||
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
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user