add packageing module

2025-11-10 14:49:00 +02:00
parent a18ad541f2
commit 9d0cc5b221
6 changed files with 847 additions and 150 deletions
--- a/UI_V2/dbActions/bulk_products.py
+++ b/UI_V2/dbActions/bulk_products.py
@@ -0,0 +1,127 @@
+import sqlite3
+from typing import Optional
+
+class BulkProducts:
+    def __init__(self, db_path="instance/app_database.db"):
+        self.db_path = db_path
+        self._create_bulk_products_table()
+    
+    def _create_bulk_products_table(self):
+        with sqlite3.connect(self.db_path) as conn:
+            cursor = conn.cursor()
+            cursor.execute("""
+                CREATE TABLE IF NOT EXISTS bulk_products (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    bill_id INTEGER,
+                    name TEXT,
+                    mesure_unit TEXT,
+                    quantity DOUBLE,
+                    price DOUBLE,
+                    vat INTEGER default 0,
+                    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+                );
+            """)
+            conn.commit()
+
+    def add_bulk_products_product(self, bill_id, name, mesure_unit, quantity, price, vat):
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                cursor.execute("""
+                    INSERT INTO bulk_products (bill_id, name, mesure_unit, quantity, price, vat) 
+                    VALUES (?, ?, ?, ?, ?, ?)
+                """, (bill_id, name, mesure_unit, quantity, price, vat))
+                conn.commit()
+            return True
+        except sqlite3.IntegrityError:
+            return False
+
+    def get_all(self):
+        with sqlite3.connect(self.db_path) as conn:
+            cursor = conn.cursor()
+            cursor.execute("""
+                SELECT * FROM bulk_products 
+            """,)
+            rows = cursor.fetchall()
+            result = []
+            if rows:
+                for row in rows: 
+                    buffer =  {
+                        "id": row[0], 
+                        "bill_id": row[1], 
+                        "name": row[2],
+                        "mesure_unit": row[3],
+                        "quantity": row[4],
+                        "price": row[5],
+                        "vat": row[6],
+                        "created_at": row[7]
+                    }
+                    result.append(buffer)
+                return result
+            return []
+    
+    def get_one(self, id):
+        with sqlite3.connect(self.db_path) as conn:
+            cursor = conn.cursor()
+            cursor.execute("""
+                SELECT * FROM bulk_products WHERE id = ?
+            """,(id, ))
+            row = cursor.fetchone()
+            result = []
+            if row:
+                result = {
+                        "id": row[0], 
+                        "bill_id": row[1], 
+                        "name": row[2],
+                        "mesure_unit": row[3],
+                        "quantity": row[4],
+                        "price": row[5],
+                        "vat": row[6],
+                        "created_at": row[7]
+                    }
+                return result
+            return None
+        
+    def get_product_by_bill_id(self, bill_id):
+        with sqlite3.connect(self.db_path) as conn:
+            cursor = conn.cursor()
+            cursor.execute("""
+                SELECT * FROM bulk_products WHERE bill_id = ?
+            """,(bill_id, ))
+            row = cursor.fetchone()
+            result = []
+            if row:
+                result = {
+                        "id": row[0], 
+                        "bill_id": row[1], 
+                        "name": row[2],
+                        "mesure_unit": row[3],
+                        "quantity": row[4],
+                        "price": row[5],
+                        "vat": row[6],
+                        "created_at": row[7]
+                    }
+                return result
+            return None
+        
+    def remove(self, id):
+         with sqlite3.connect(self.db_path) as conn:
+            cursor = conn.cursor()
+            cursor.execute('''
+                DELETE FROM bulk_products WHERE id=?;
+            ''', (id,))
+            conn.commit()
+
+    def update(self, id, bill_id, name, mesure_unit, quantity, price, vat):
+        try:
+            with sqlite3.connect(self.db_path) as conn:
+                cursor = conn.cursor()
+                cursor.execute("""
+                    UPDATE bulk_products
+                    SET bill_id = ?, name = ?, mesure_unit = ?, quantity = ?, price = ?, vat = ?
+                    WHERE id = ?
+                """, (bill_id, name, mesure_unit, quantity, price, vat, id))
+                conn.commit()
+            return True
+        except Exception:
+            return False