README.md

@@ -2,9 +2,3 @@
 
 ** 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

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

@@ -1,101 +0,0 @@
-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

@@ -1,66 +0,0 @@
-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

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