moved files into layer dir

merged branch layer_interface_adapters into layer_framework_driver_db

.gitignore

@@ -0,0 +1,4 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class

README.md

@@ -3,8 +3,26 @@
 ** Master MDS HSMA ** 
 Clean Architecture - Architekturstil am praktischen Beispiel z.B. in Python. 
 
+run python script as a module e.g.: python -m framework_driver.db.db_cart
 
 
-
-# Layer: Framework Driver Database
-- Implemented a small sqllite3 database for a online shop
+## Layers
+- entities
+    - cart_item.py
+    - cart.py
+    - product.py
+- use_cases
+    - cart_reporsitory_interface.py
+    - view_cart.py
+- interfaces_adapters
+    - cart_database_interface.py
+    - cart_repository.py
+- framework_driver
+    - db
+        - db_cart.py
+        - db_product.py
+        - db_setup.py
+        - db_user.py
+        - shop.db
+    - ui
+        - coming soon ...

src/entities/cart.py

@@ -0,0 +1,27 @@
+#python imports
+from typing import List
+from dataclasses import dataclass
+
+#dependency imports
+from entities.cart_item import CartItem
+from entities.product import Product
+
+@dataclass
+class Cart:
+    items: List[CartItem]
+
+    def add_item(self, product: Product, quantity: int):
+        for item in self.items:
+            if item.product.id == product.id:
+                item.quantity += quantity
+                return
+        self.items.append(CartItem(product=product, quantity=quantity))
+
+    def remove_item(self, product_id: str):
+        self.items = [item for item in self.items if item.product.id != product_id]
+
+    def calculate_total_price(self) -> float:
+        return sum(item.calculate_total_price() for item in self.items)
+
+    def list_items(self) -> List[str]:
+        return [f"{item.quantity} x {item.product.name} - {item.calculate_total_price()} EUR" for item in self.items]

src/entities/cart_item.py

@@ -0,0 +1,17 @@
+#python imports
+from typing import List
+from dataclasses import dataclass
+
+@dataclass
+class CartItem:
+    product_id: int
+    name: str
+    quantity: int
+    price: float
+
+    def post_init(self):
+        if self.quantity <= 0:
+            raise ValueError("Quantity has to be atleast 1")
+
+    def calculate_total_price(self) -> float:
+        return self.quantity * self.price

src/entities/product.py

@@ -0,0 +1,13 @@
+#python imports
+from dataclasses import dataclass
+
+@dataclass
+class Product:
+    id: str
+    name: str
+    description: str
+    price: float
+
+    def post_init(self):
+        if self.price < 0:
+            raise ValueError("Der Preis darf nicht negativ sein.")

src/framework_driver/db/db_cart.py

@@ -1,6 +1,8 @@
 import sqlite3
+from typing import Optional, List
+from interface_adapters.cart_database_interface import CartDatabaseInterface
 
-class CartDatabase:
+class CartDatabase(CartDatabaseInterface):
     def __init__(self, db_file):
         """Initialize the CartDatabase with a connection to the SQLite database."""
         self.conn = sqlite3.connect(db_file)
@@ -16,16 +18,17 @@ class CartDatabase:
         except sqlite3.Error as e:
             print(e)
 
-    def get_cart_for_user(self, user_id):
-        """Retrieve the cart items for a specific user."""
+    def fetch_cart_items(self, user_id: int) -> Optional[List[dict]]:
+        """Fetch cart items from the database for a given user ID."""
         sql = "SELECT * FROM cart WHERE user_id = ?"
         try:
             c = self.conn.cursor()
             c.execute(sql, (user_id,))
-            return c.fetchall()
+            rows = c.fetchall()
+            return [{"id": row[0], "user_id": row[1], "product_id": row[2], "quantity": row[3]} for row in rows]
         except sqlite3.Error as e:
             print(e)
-            return []
+            return None
 
     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."""
@@ -57,19 +60,9 @@ class CartDatabase:
         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"
+    database = "framework_driver/db/shop.db"
 
     cart_db = CartDatabase(database)
 
@@ -81,14 +74,11 @@ def main():
     # 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))
+    print(f"Cart items for user {ex_user_id}:", cart_db.fetch_cart_items(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))
+    print(f"Cart items for user {ex_user_id}:", cart_db.fetch_cart_items(ex_user_id))
 
 if __name__ == "__main__":
     main()

src/framework_driver/db/db_product.py

@@ -59,7 +59,7 @@ class ProductDatabase:
             print(e)
 
 def main():
-    database = "db/shop.db"
+    database = "framework_driver/db/shop.db"
 
     product_db = ProductDatabase(database)
 

src/framework_driver/db/db_setup.py

@@ -29,7 +29,7 @@ def delete_table(conn, table_name):
         print(e)
 
 def main():
-    database = "db/shop.db"
+    database = "framework_driver/db/shop.db"
 
     sql_user_table = """CREATE TABLE IF NOT EXISTS users (
         id INTEGER PRIMARY KEY,

src/framework_driver/db/db_user.py

@@ -60,7 +60,7 @@ class UserDatabase:
             return []
 
 def main():
-    database = "db/shop.db"
+    database = "framework_driver/db/shop.db"
 
     user_db = UserDatabase(database)
 

src/interface_adapters/cart_database_interface.py

@@ -0,0 +1,8 @@
+from typing import Optional, List
+
+class CartDatabaseInterface:
+    def fetch_cart_items(self, user_id: int) -> Optional[List[dict]]:
+        """
+        Abstract method to fetch cart items from the database for a given user ID.
+        """
+        raise NotImplementedError("fetch_cart_items must be implemented by a subclass")

src/interface_adapters/cart_repository.py

@@ -0,0 +1,13 @@
+#python imports
+from typing import Optional, List
+
+#dependency imports
+from use_cases.cart_repository_interface import CartRepositoryInterface
+from interface_adapters.cart_database_interface import CartDatabaseInterface
+
+class CartRepository(CartRepositoryInterface):
+    def __init__(self, database_gateway: CartDatabaseInterface):
+        self.database_gateway = database_gateway
+
+    def fetch_cart_by_user_id(self, user_id: int) -> Optional[List[dict]]:
+        return self.database_gateway.fetch_cart_items(user_id)

src/use_cases/cart_repository_interface.py

@@ -0,0 +1,8 @@
+from typing import Optional, List
+
+class CartRepositoryInterface:
+    def fetch_cart_by_user_id(self, user_id: int) -> Optional[List[dict]]:
+        """
+        Abstract method to fetch cart data by user ID.
+        """
+        raise NotImplementedError("fetch_cart_by_user_id must be implemented by a subclass")

src/use_cases/view_cart.py

@@ -0,0 +1,22 @@
+#python imports
+from typing import Optional
+
+#dependency imports
+from entities.cart import Cart, CartItem
+from interface_adapters.cart_repository import CartRepository
+
+class ViewCart:
+    def __init__(self, cart_repository: CartRepository):
+        self.cart_repository = cart_repository
+
+    def execute(self, user_id: int) -> Optional[Cart]:
+        """
+        Fetches the cart data from the repository, converts it to Cart entity.
+        """
+        cart_data = self.cart_repository.fetch_cart_by_user_id(user_id)
+        if not cart_data:
+            return None
+
+        # Convert raw data to domain entities
+        items = [CartItem(**item) for item in cart_data]
+        return Cart(items)