94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
|
import sqlite3
|
||
|
|
|
||
|
|
class CartDatabase:
|
||
|
|
def __init__(self, db_file):
|
||
|
|
"""Initialize the CartDatabase with a connection to the SQLite database."""
|
||
|
|
self.conn = sqlite3.connect(db_file)
|
||
|
|
|
||
|
|
def add_to_cart(self, user_id, product_id, quantity):
|
||
|
|
"""Add a product to the user's cart."""
|
||
|
|
sql = "INSERT INTO cart (user_id, product_id, quantity) VALUES (?, ?, ?)"
|
||
|
|
try:
|
||
|
|
c = self.conn.cursor()
|
||
|
|
c.execute(sql, (user_id, product_id, quantity))
|
||
|
|
self.conn.commit()
|
||
|
|
return c.lastrowid
|
||
|
|
except sqlite3.Error as e:
|
||
|
|
print(e)
|
||
|
|
|
||
|
|
def get_cart_for_user(self, user_id):
|
||
|
|
"""Retrieve the cart items for a specific user."""
|
||
|
|
sql = "SELECT * FROM cart WHERE user_id = ?"
|
||
|
|
try:
|
||
|
|
c = self.conn.cursor()
|
||
|
|
c.execute(sql, (user_id,))
|
||
|
|
return c.fetchall()
|
||
|
|
except sqlite3.Error as e:
|
||
|
|
print(e)
|
||
|
|
return []
|
||
|
|
|
||
|
|
def update_cart_item(self, user_id, product_id, quantity):
|
||
|
|
"""Update the quantity of a specific cart item based on user_id and product_id."""
|
||
|
|
sql = "UPDATE cart SET quantity = ? WHERE user_id = ? AND product_id = ?"
|
||
|
|
try:
|
||
|
|
c = self.conn.cursor()
|
||
|
|
c.execute(sql, (quantity, user_id, product_id))
|
||
|
|
self.conn.commit()
|
||
|
|
except sqlite3.Error as e:
|
||
|
|
print(e)
|
||
|
|
|
||
|
|
def remove_from_cart(self, cart_id):
|
||
|
|
"""Remove a specific item from the cart."""
|
||
|
|
sql = "DELETE FROM cart WHERE id = ?"
|
||
|
|
try:
|
||
|
|
c = self.conn.cursor()
|
||
|
|
c.execute(sql, (cart_id,))
|
||
|
|
self.conn.commit()
|
||
|
|
except sqlite3.Error as e:
|
||
|
|
print(e)
|
||
|
|
|
||
|
|
def clear_cart(self, user_id):
|
||
|
|
"""Clear all items from a user's cart."""
|
||
|
|
sql = "DELETE FROM cart WHERE user_id = ?"
|
||
|
|
try:
|
||
|
|
c = self.conn.cursor()
|
||
|
|
c.execute(sql, (user_id,))
|
||
|
|
self.conn.commit()
|
||
|
|
except sqlite3.Error as e:
|
||
|
|
print(e)
|
||
|
|
|
||
|
|
def get_cart_total(self, user_id):
|
||
|
|
"""Calculate the total cost of all items in a user's cart."""
|
||
|
|
sql = "SELECT SUM(products.price * cart.quantity) FROM cart JOIN products ON cart.product_id = products.id WHERE cart.user_id = ?"
|
||
|
|
try:
|
||
|
|
c = self.conn.cursor()
|
||
|
|
c.execute(sql, (user_id,))
|
||
|
|
return c.fetchone()[0]
|
||
|
|
except sqlite3.Error as e:
|
||
|
|
print(e)
|
||
|
|
return 0
|
||
|
|
|
||
|
|
def main():
|
||
|
|
database = "db/shop.db"
|
||
|
|
|
||
|
|
cart_db = CartDatabase(database)
|
||
|
|
|
||
|
|
ex_user_id = 1
|
||
|
|
|
||
|
|
# delete all cart items
|
||
|
|
cart_db.clear_cart(ex_user_id)
|
||
|
|
|
||
|
|
# user id, product id, quantity
|
||
|
|
cart_db.add_to_cart(ex_user_id, 1, 2)
|
||
|
|
cart_db.add_to_cart(ex_user_id, 3, 5)
|
||
|
|
|
||
|
|
print(f"Cart items for user {ex_user_id}:", cart_db.get_cart_for_user(ex_user_id))
|
||
|
|
print(f"Total cost for user {ex_user_id}:", cart_db.get_cart_total(ex_user_id))
|
||
|
|
|
||
|
|
# user id, product id, quantity
|
||
|
|
cart_db.update_cart_item(ex_user_id, 3, 3)
|
||
|
|
print(f"Cart items for user {ex_user_id}:", cart_db.get_cart_for_user(ex_user_id))
|
||
|
|
print(f"Total cost for user {ex_user_id}:", cart_db.get_cart_total(ex_user_id))
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|