added description

README.md

@@ -2,3 +2,9 @@
 
 ** Master MDS HSMA ** 
 Clean Architecture - Architekturstil am praktischen Beispiel z.B. in Python. 
+
+
+
+
+# Layer: Framework Driver Database
+- Implemented a small sqllite3 database for a online shop

db/db_cart.py

@@ -0,0 +1,94 @@
+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()

db/db_product.py

@@ -0,0 +1,101 @@
+import sqlite3
+
+class ProductDatabase:
+    def __init__(self, db_file):
+        """Initialize the ProductDatabase with a connection to the SQLite database."""
+        self.conn = sqlite3.connect(db_file)
+
+    def add_product(self, product):
+        """Add a new product to the products table."""
+        sql = "INSERT INTO products (name, price, quantity) VALUES (?, ?, ?)"
+        try:
+            c = self.conn.cursor()
+            c.execute(sql, product)
+            self.conn.commit()
+            return c.lastrowid
+        except sqlite3.Error as e:
+            print(e)
+
+    def get_product_by_id(self, product_id):
+        """Retrieve a product by its ID."""
+        sql = "SELECT * FROM products WHERE id = ?"
+        try:
+            c = self.conn.cursor()
+            c.execute(sql, (product_id,))
+            return c.fetchone()
+        except sqlite3.Error as e:
+            print(e)
+            return None
+
+    def get_all_products(self):
+        """Retrieve all products."""
+        sql = "SELECT * FROM products"
+        try:
+            c = self.conn.cursor()
+            c.execute(sql)
+            return c.fetchall()
+        except sqlite3.Error as e:
+            print(e)
+            return []
+
+    def update_product(self, product_id, product_data):
+        """Update a product's information."""
+        sql = "UPDATE products SET name = ?, price = ?, quantity = ? WHERE id = ?"
+        try:
+            c = self.conn.cursor()
+            c.execute(sql, (*product_data, product_id))
+            self.conn.commit()
+        except sqlite3.Error as e:
+            print(e)
+
+    def delete_product(self, product_id):
+        """Delete a product by its ID."""
+        sql = "DELETE FROM products WHERE id = ?"
+        try:
+            c = self.conn.cursor()
+            c.execute(sql, (product_id,))
+            self.conn.commit()
+        except sqlite3.Error as e:
+            print(e)
+
+def main():
+    database = "db/shop.db"
+
+    product_db = ProductDatabase(database)
+
+    # delete all products
+    for product in product_db.get_all_products():
+        product_db.delete_product(product[0])
+
+    # name, price, quantity
+    products = [("Apel", 2.5, 5),
+                ("Banana", 1.5, 10),
+                ("Orange", 3.2, 7)]
+
+    for product in products:
+        product_db.add_product(product)
+
+    products = product_db.get_all_products()
+    print("All products:")
+    for product in products:
+        print(product)
+
+    # name, price, quantity
+    product_data = ("Orange", 3.0, 5)
+    product_db.update_product(3, product_data)
+
+    product = product_db.get_product_by_id(1)
+    print("Updated product:")
+    print(product)
+
+    product_db.delete_product(2)
+    print("Product deleted")
+
+    products = product_db.get_all_products()
+    print("All products:")
+    for product in products:
+        print(product)
+
+
+if __name__ == "__main__":
+    main()

db/db_setup.py

@@ -0,0 +1,66 @@
+import sqlite3
+
+def create_connection(db_file):
+    """Create a database connection to the SQLite database specified by db_file."""
+    conn = None
+    try:
+        conn = sqlite3.connect(db_file)
+        print(f"Connected to {db_file}")
+    except sqlite3.Error as e:
+        print(e)
+    return conn
+
+def create_table(conn, create_table_sql):
+    """Create a table from the create_table_sql statement."""
+    try:
+        c = conn.cursor()
+        c.execute(create_table_sql)
+        print("Table created successfully")
+    except sqlite3.Error as e:
+        print(e)
+
+def delete_table(conn, table_name):
+    """Delete a table specified by table_name."""
+    try:
+        c = conn.cursor()
+        c.execute(f"DROP TABLE IF EXISTS {table_name}")
+        print(f"Table {table_name} deleted successfully")
+    except sqlite3.Error as e:
+        print(e)
+
+def main():
+    database = "db/shop.db"
+
+    sql_user_table = """CREATE TABLE IF NOT EXISTS users (
+        id INTEGER PRIMARY KEY,
+        name TEXT NOT NULL,
+        email TEXT NOT NULL
+    )"""
+
+    sql_product_table = """CREATE TABLE IF NOT EXISTS products (
+        id INTEGER PRIMARY KEY,
+        name TEXT NOT NULL,
+        price REAL NOT NULL,
+        quantity INTEGER NOT NULL
+    )"""
+
+    sql_cart_table = """CREATE TABLE IF NOT EXISTS cart (
+        id INTEGER PRIMARY KEY,
+        user_id INTEGER NOT NULL,
+        product_id INTEGER NOT NULL,
+        quantity INTEGER NOT NULL,
+        FOREIGN KEY (user_id) REFERENCES users (id),
+        FOREIGN KEY (product_id) REFERENCES products (id)
+    )"""
+
+    conn = create_connection(database)
+
+    if conn is not None:
+        create_table(conn, sql_user_table)
+        create_table(conn, sql_product_table)
+        create_table(conn, sql_cart_table)
+    else:
+        print("Error! Cannot create the database connection.")
+
+if __name__ == '__main__':
+    main()

db/db_user.py

@@ -0,0 +1,89 @@
+import sqlite3
+
+class UserDatabase:
+    def __init__(self, db_file):
+        """Initialize the UserDatabase with a connection to the SQLite database."""
+        self.conn = sqlite3.connect(db_file)
+
+    def add_user(self, user):
+        """Add a new user to the users table and return the new user's id."""
+        sql = "INSERT INTO users (name, email) VALUES (?, ?)"
+        try:
+            c = self.conn.cursor()
+            c.execute(sql, user)
+            self.conn.commit()
+            return c.lastrowid
+        except sqlite3.Error as e:
+            print(e)
+            return None
+
+    def get_user_by_id(self, user_id):
+        """Retrieve a user by their ID."""
+        sql = "SELECT * FROM users WHERE id = ?"
+        try:
+            c = self.conn.cursor()
+            c.execute(sql, (user_id,))
+            return c.fetchone()
+        except sqlite3.Error as e:
+            print(e)
+            return None
+
+    def update_user(self, user_id, user_data):
+        """Update a user's information."""
+        sql = "UPDATE users SET name = ?, email = ? WHERE id = ?"
+        try:
+            c = self.conn.cursor()
+            c.execute(sql, (*user_data, user_id))
+            self.conn.commit()
+        except sqlite3.Error as e:
+            print(e)
+
+    def delete_user(self, user_id):
+        """Delete a user by their ID."""
+        sql = "DELETE FROM users WHERE id = ?"
+        try:
+            c = self.conn.cursor()
+            c.execute(sql, (user_id,))
+            self.conn.commit()
+        except sqlite3.Error as e:
+            print(e)
+    
+    def get_all_users(self):
+        """Retrieve all users."""
+        sql = "SELECT * FROM users"
+        try:
+            c = self.conn.cursor()
+            c.execute(sql)
+            return c.fetchall()
+        except sqlite3.Error as e:
+            print(e)
+            return []
+
+def main():
+    database = "db/shop.db"
+
+    user_db = UserDatabase(database)
+
+    # delete all users
+    for user in user_db.get_all_users():
+        user_db.delete_user(user[0])
+
+    users = [("Michael", "michael@mail.com"),
+             ("Klöara", "klara@mail.com"),
+             ("Arman", "arman@mail.com")]
+
+    for user in users:
+        user_id = user_db.add_user(user)
+
+    user = user_db.get_user_by_id(1)
+    print(f"Retrieved user: {user}")
+
+    user_db.update_user(2, ("Klara", "klara@mail.com"))
+
+    users = user_db.get_all_users()
+    for user in users:
+        print(user)
+
+
+if __name__ == "__main__":
+    main()