Merge pull request #4 from Abishevs/orm-based-db

Orm based db manager with SQLalchemy

Author: Abishevs

src/tech_cache/commons/database_manager.py

@@ -1,157 +1,67 @@
-import sqlite3
 from typing import List
-from tech_cache.models.item import Item
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.exc import SQLAlchemyError
+from tech_cache.models.item import Base, Item
 
 class DatabaseManager:
     """Handles database connection and data persitance
     Idea is that an instance of this class is passed to
     models, so that views can then interpret and draw
     data.
     """
-    def __init__(self):
-        self.con: sqlite3.Connection
-        self.cur: sqlite3.Cursor
-        self.db_name = "test.db"
-        self.connect()
+    def __init__(self, db_url="sqlite:///test.db"):
+        self.engine = create_engine(db_url, echo=True)
+        Base.metadata.create_all(self.engine) 
+        self.Session = sessionmaker(bind=self.engine)
 
-    def connect(self):
-        """Open db connection when you open application"""
-        self.con = sqlite3.connect(self.db_name)
-        self.cur = self.con.cursor()
-        self.init_db()
+    def get_session(self):
+        return self.Session()
 
-    def close(self):
-        """close db before closing application"""
-        self.con.close()
-
-    def init_db(self):
-        create_table_query = """
-        CREATE TABLE IF NOT EXISTS items (
-                id TEXT PRIMARY KEY,
-                name TEXT,
-                category TEXT,
-                quantity INTEGER,
-                sku TEXT,
-                specification TEXT,
-                created_at TEXT,
-                updated_at TEXT
-        );
-        """
-        self.execute_query(create_table_query)
-
-    def execute_query(self, query:str, data: tuple | None=None):
-        with self.con:
-            if data:
-                self.cur.execute(query, data)
-            else:
-                self.cur.execute(query)
-
-    def get_item(self, uid):
-        query = """
-        SELECT 
-                id,
-                name, 
-                category,
-                quantity,
-                sku, 
-                specification,
-                created_at,
-                updated_at
-        FROM items
-        WHERE id = ?;
-        """
-        self.execute_query(query, (uid,))
-        db_item = self.cur.fetchone()
-        new_item = Item(
-                    id=db_item[0],
-                    name=db_item[1],
-                    category=db_item[2],
-                    quantity=db_item[3],
-                    sku=db_item[4],
-                    specification=db_item[5],
-                    created_at=db_item[6],
-                    updated_at=db_item[7]
-                        )
-        return new_item
+    def get_item(self, item_id:int):
+        with self.get_session() as session:
+            return session.query(Item).filter(Item.id == item_id).one_or_none()
 
     def add_item(self, item: Item):
-        query = """
-        INSERT INTO items (
-                id,
-                name, 
-                category,
-                quantity,
-                sku, 
-                specification,
-                created_at,
-                updated_at
-                ) 
-        VALUES(?, ?, ?, ?, ?, ?, ?, ?);
-        """
-        data = (str(item.id),
-                item.name,
-                item.category,
-                item.quantity,
-                item.sku,
-                item.specification,
-                item.created_at,
-                item.updated_at,
-                )
-        self.execute_query(query, data)
+        with self.get_session() as session:
+            try:
+                session.add(item)
+                session.commit()
+            except SQLAlchemyError as e:
+                session.rollback()
+                raise e
 
-    def update_item(self, item: Item):
-        query = """
-        UPDATE items 
-        SET name = ?,
-            category = ?,
-            quantity = ?,
-            sku = ?, 
-            specification = ?,
-            updated_at = ?
-        WHERE id = ?;
-        """
-        data = (item.name, 
-                item.category, 
-                item.quantity,
-                item.sku,
-                item.specification,
-                item.updated_at,
+    def update_item(self, item_id: int):
 
-                item.id)
-        self.execute_query(query, data)
+        with self.get_session() as session:
+            try:
+                item = session.query(Item).filter(Item.id == item_id).one_or_none()
+                
+                if item:
+                    session.delete(item)
+                    session.commit()
+                else:
+                    print(f"Item with ID {item_id} not found.")
+            except SQLAlchemyError as e:
+                session.rollback()
+                raise e
 
-    def delete_item(self, uid:str):
-        query = "DELETE FROM items WHERE id = ?;"
-        self.execute_query(query, (uid,))
+    def delete_item(self, item_id: int):
+        session = self.get_session()
+        with self.get_session() as session:
+            try:
+                item = session.query(Item).filter(Item.id == item_id).one_or_none()
+                
+                if item:
+                    session.delete(item)
+                    session.commit()
+                else:
+                    print(f"Item with ID {item_id} not found.")
+
+            except SQLAlchemyError as e:
+                session.rollback()
+                raise e
 
     def get_all_items(self) -> List[Item]:
-        query = """
-        SELECT 
-                id,
-                name, 
-                category,
-                quantity,
-                sku, 
-                specification,
-                created_at,
-                updated_at
-        FROM items;
-        """
-        self.execute_query(query)
-        db_items = self.cur.fetchall()
-        items = []
-        for row in db_items:
-            # follows query parameter order
-            new_item = Item(
-                    id=row[0],
-                    name=row[1],
-                    category=row[2],
-                    quantity=row[3],
-                    sku=row[4],
-                    specification=row[5],
-                    created_at=row[6],
-                    updated_at=row[7]
-                        )
-            items.append(new_item)
-        return items
-
+        with self.get_session() as session:
+            return list(session.query(Item))

src/tech_cache/models/item.py

@@ -1,54 +1,46 @@
-from typing import Any, List
+from typing import Any
+from typing import List
+from typing import Optional
 from datetime import datetime
-from uuid import uuid4
+from sqlalchemy import String
+from sqlalchemy import DateTime
+from sqlalchemy import func
+from sqlalchemy import String
+from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
 
-class Item:
-    """Represents rows in tableview """
-    def __init__(self, **kwargs): 
-        self.id:str = str(kwargs.get('id', uuid4())) # for in database  data integrity
-        self._name:str = kwargs.get('name', "")
-        self.category:str = kwargs.get('category', "")
-        self._quantity:int = kwargs.get('quantity', 0)
-        self.sku:str = kwargs.get('sku', "") # TODO: Gen SKU
-        self.specification:str = kwargs.get('specification', "")
-        self.created_at:str = kwargs.get('created_at', datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
-        self.updated_at:str = kwargs.get('updated_at',datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
+class Base(DeclarativeBase):
+    pass
 
-    @property
-    def name(self):
-        """The name property."""
-        return self._name
+class Item(Base):
+    __tablename__ = "items"
 
-    @name.setter
-    def name(self, value):
-        if value == "":
-            raise ValueError("Mandatory name field can't be empty")
-        self._name = value
+    id: Mapped[int] = mapped_column(primary_key=True)
+    sku: Mapped[str] = mapped_column(String(30))
+    name: Mapped[str] = mapped_column(String(30))
+    category: Mapped[str] = mapped_column(String(30))
+    quantity: Mapped[int] = mapped_column()
+    specification: Mapped[Optional[str]] = mapped_column(String(255))
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(DateTime, default=func.now(), onupdate=func.now())
 
-    @property
-    def quantity(self):
-        """The quantity property."""
-        return self._quantity
+    def __init__(self, 
+                 name:str = "",
+                 sku:str = "",
+                 category:str = "",
+                 quantity:int = 0,
+                 specification:str = "",
+                 ):
 
-    @quantity.setter
-    def quantity(self, value:int):
-        if not isinstance(value, int):
-            raise ValueError("Quantity must be an integer")
+        self.name = name
+        self.sku = sku
+        self.category = category
+        self.quantity = quantity
+        self.specification = specification
 
-        if value < 0:
-            raise ValueError("Quantity cannot be negative")
-
-        self._quantity = value
-
-    def get_fields(self) -> List[Any]:
-        """Item fields to display"""
-        fields = []
-        fields.append(self.sku)
-        fields.append(self.name)
-        fields.append(self.category)
-        fields.append(self.quantity)
-        fields.append(self.specification)
-        return fields
+    def __repr__(self) -> str:
+        return f"Item(name={self.name!r}, category={self.category!r}, quantity={self.quantity!r})"
 
     def __getitem__(self, column_index):
         if column_index == 0:
@@ -64,4 +56,12 @@ def __getitem__(self, column_index):
         else:
             raise IndexError("Invalid column index")
 
-        
+    def get_fields(self) -> List[Any]:
+        """Item fields to display"""
+        fields = []
+        fields.append(self.sku)
+        fields.append(self.name)
+        fields.append(self.category)
+        fields.append(self.quantity)
+        fields.append(self.specification)
+        return fields

src/tech_cache/views/main_window.py

@@ -174,7 +174,6 @@ def closeEvent(self, event):
 
         if reply == QtWidgets.QMessageBox.StandardButton.Yes:
             # TODO: clean up
-            self.database.close()
             event.accept()
         else:
             event.ignore()