186 lines
6.2 KiB
Python
186 lines
6.2 KiB
Python
import sqlite3
|
|
from typing import Optional
|
|
|
|
class Orders:
|
|
def __init__(self, db_path="instance/app_database.db"):
|
|
self.db_path = db_path
|
|
self._create_orders_table()
|
|
self._create_orders_map_table()
|
|
|
|
def _create_orders_table(self):
|
|
"""Create the orders table if it doesn't already exist."""
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS orders (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id TEXT NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'on_hold'
|
|
);
|
|
""")
|
|
conn.commit()
|
|
|
|
def _create_orders_map_table(self):
|
|
"""Create the orders table if it doesn't already exist."""
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS orders_products (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
product_id TEXT,
|
|
orders_id TEXT,
|
|
quantity INTEGER
|
|
);
|
|
|
|
""")
|
|
conn.commit()
|
|
|
|
def add_order(self, user_id):
|
|
try:
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT INTO orders (user_id)
|
|
VALUES (?)
|
|
""", (user_id,))
|
|
conn.commit()
|
|
return True
|
|
except sqlite3.IntegrityError:
|
|
return False
|
|
|
|
def add_product_to_order(self, product_id, orders_id, quantity):
|
|
try:
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT INTO orders_products (product_id, orders_id, quantity)
|
|
VALUES (?,?,?)
|
|
""", (product_id, orders_id, quantity))
|
|
conn.commit()
|
|
return True
|
|
except sqlite3.IntegrityError:
|
|
return False
|
|
|
|
def get_on_hold_order(self, user_id):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT * FROM orders
|
|
WHERE user_id = ? and status = 'on_hold'
|
|
""", (user_id,))
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return {
|
|
"id": row[0],
|
|
"user_id": row[1],
|
|
"status": row[2],
|
|
}
|
|
return None
|
|
|
|
def update_order_status(self, status, id):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
UPDATE orders SET status = ?
|
|
WHERE id = ?
|
|
''', (status, id))
|
|
conn.commit()
|
|
|
|
def update_order_user_id(self, user_id, id):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
UPDATE orders SET user_id = ?
|
|
WHERE id = ?
|
|
''', (user_id, id))
|
|
conn.commit()
|
|
|
|
def get_order_products(self, id):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT * FROM orders_products
|
|
WHERE orders_id = ?
|
|
""", (id,))
|
|
rows = cursor.fetchall()
|
|
result = []
|
|
if rows:
|
|
for row in rows:
|
|
buffer = {
|
|
"id": row[0],
|
|
"prdouct_id": row[1],
|
|
"orders_id": row[2],
|
|
"quantity": row[3]
|
|
}
|
|
result.append(buffer)
|
|
return result
|
|
return []
|
|
|
|
def update_order_map_quantity(self, id, quantity):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
UPDATE orders_products SET quantity = ?
|
|
WHERE id = ?
|
|
''', (quantity, id))
|
|
conn.commit()
|
|
|
|
def get_orders(self):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT * FROM orders
|
|
""")
|
|
rows = cursor.fetchall()
|
|
if rows:
|
|
buffer = []
|
|
for row in rows:
|
|
r = {
|
|
"id": row[0],
|
|
"user_id": row[1],
|
|
"status": row[2],
|
|
}
|
|
buffer.append(r)
|
|
return buffer
|
|
return []
|
|
|
|
def get_orders_for_user(self, user_id):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT * FROM orders WHERE user_id = ?
|
|
""", (user_id,))
|
|
rows = cursor.fetchall()
|
|
if rows:
|
|
buffer = []
|
|
for row in rows:
|
|
r = {
|
|
"id": row[0],
|
|
"user_id": row[1],
|
|
"status": row[2],
|
|
}
|
|
buffer.append(r)
|
|
return buffer
|
|
return []
|
|
|
|
def remove_product_from_order(self, order_id, product_id):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
DELETE FROM orders_products WHERE orders_id=? and product_id=?;
|
|
''', (order_id, product_id))
|
|
conn.commit()
|
|
|
|
def remove_order(self, order_id):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
DELETE FROM orders_products WHERE orders_id=?;
|
|
''', (order_id,))
|
|
conn.commit()
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
DELETE FROM orders WHERE id=?;
|
|
''', (order_id,))
|
|
conn.commit() |