add V2
This commit is contained in:
BIN
UI_V2/dbActions/__pycache__/categories.cpython-313.pyc
Normal file
BIN
UI_V2/dbActions/__pycache__/categories.cpython-313.pyc
Normal file
Binary file not shown.
BIN
UI_V2/dbActions/__pycache__/company.cpython-313.pyc
Normal file
BIN
UI_V2/dbActions/__pycache__/company.cpython-313.pyc
Normal file
Binary file not shown.
BIN
UI_V2/dbActions/__pycache__/orders.cpython-313.pyc
Normal file
BIN
UI_V2/dbActions/__pycache__/orders.cpython-313.pyc
Normal file
Binary file not shown.
BIN
UI_V2/dbActions/__pycache__/products.cpython-313.pyc
Normal file
BIN
UI_V2/dbActions/__pycache__/products.cpython-313.pyc
Normal file
Binary file not shown.
BIN
UI_V2/dbActions/__pycache__/users.cpython-313.pyc
Normal file
BIN
UI_V2/dbActions/__pycache__/users.cpython-313.pyc
Normal file
Binary file not shown.
94
UI_V2/dbActions/categories.py
Normal file
94
UI_V2/dbActions/categories.py
Normal file
@@ -0,0 +1,94 @@
|
||||
import sqlite3
|
||||
from typing import Optional
|
||||
|
||||
class Categories:
|
||||
def __init__(self, db_path="instance/app_database.db"):
|
||||
self.db_path = db_path
|
||||
self._create_categories_table()
|
||||
|
||||
def _create_categories_table(self):
|
||||
"""Create the users 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 categories (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
name TEXT,
|
||||
image TEXT,
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
status TEXT NOT NULL DEFAULT 'active'
|
||||
);
|
||||
|
||||
""")
|
||||
conn.commit()
|
||||
|
||||
def add(self, name, image: str) -> bool:
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
INSERT INTO categories (name, image)
|
||||
VALUES (?, ?)
|
||||
""", (name, image))
|
||||
conn.commit()
|
||||
return True
|
||||
except sqlite3.IntegrityError:
|
||||
return False
|
||||
|
||||
def get(self, id):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT * FROM categories
|
||||
WHERE id = ?
|
||||
""", (id,))
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return {
|
||||
"id": row[0],
|
||||
"name": row[1],
|
||||
"image": row[2],
|
||||
"created_at": row[3],
|
||||
"status":row[4],
|
||||
}
|
||||
else:
|
||||
return None
|
||||
|
||||
def get_categories(self):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT * FROM categories;
|
||||
""")
|
||||
rows = cursor.fetchall()
|
||||
if rows:
|
||||
buffer = []
|
||||
for row in rows:
|
||||
r = {
|
||||
"id": row[0],
|
||||
"name": row[1],
|
||||
"image": row[2],
|
||||
"created_at": row[3],
|
||||
"status":row[4],
|
||||
}
|
||||
buffer.append(r)
|
||||
return buffer
|
||||
else:
|
||||
return []
|
||||
|
||||
def update(self, name, image, id):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
UPDATE categories SET name = ?, image = ?
|
||||
WHERE id = ?
|
||||
''', (name, image, id))
|
||||
conn.commit()
|
||||
|
||||
def delete(self, id):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
DELETE FROM categories WHERE id=?;
|
||||
''', (id,))
|
||||
conn.commit()
|
||||
97
UI_V2/dbActions/company.py
Normal file
97
UI_V2/dbActions/company.py
Normal file
@@ -0,0 +1,97 @@
|
||||
import sqlite3
|
||||
from typing import Optional
|
||||
|
||||
class Company:
|
||||
def __init__(self, db_path="instance/app_database.db"):
|
||||
self.db_path = db_path
|
||||
self._create_company_table()
|
||||
|
||||
def _create_company_table(self):
|
||||
"""Create the company 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 company (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
user_id INTEGER,
|
||||
name TEXT,
|
||||
vat TEXT,
|
||||
register_number TEXT,
|
||||
address TEXT,
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
status TEXT NOT NULL DEFAULT 'active'
|
||||
);
|
||||
|
||||
""")
|
||||
conn.commit()
|
||||
|
||||
def add_company(self, company):
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
INSERT INTO company (user_id, name, vat, register_number, address)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
""", (
|
||||
company['user_id'],
|
||||
company['name'],
|
||||
company['vat'],
|
||||
company['register_number'],
|
||||
company['address']
|
||||
)
|
||||
)
|
||||
conn.commit()
|
||||
return True
|
||||
except sqlite3.IntegrityError:
|
||||
return False
|
||||
|
||||
def get_company(self, user_id):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT * FROM company
|
||||
WHERE user_id = ?
|
||||
""", (user_id,))
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return {
|
||||
'id': row[0],
|
||||
'user_id': row[1],
|
||||
'name': row[2],
|
||||
'vat': row[3],
|
||||
'register_number': row[4],
|
||||
'address': row[5],
|
||||
'created_at': row[6],
|
||||
'status': row[7],
|
||||
}
|
||||
return None
|
||||
|
||||
def update_company(self, company):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
UPDATE company SET name = ?, vat = ?, register_number = ?, address = ?
|
||||
WHERE id = ?
|
||||
''', (company['name'],
|
||||
company['vat'],
|
||||
company['register_number'],
|
||||
company['address'],
|
||||
company['id']))
|
||||
conn.commit()
|
||||
|
||||
def delete(self, id):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
DELETE FROM company WHERE id=?;
|
||||
''', (id,))
|
||||
conn.commit()
|
||||
|
||||
def deactivate(self, id):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
UPDATE company SET status = ?
|
||||
WHERE id = ?
|
||||
''', ('inactive', id))
|
||||
conn.commit()
|
||||
163
UI_V2/dbActions/orders.py
Normal file
163
UI_V2/dbActions/orders.py
Normal file
@@ -0,0 +1,163 @@
|
||||
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 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()
|
||||
140
UI_V2/dbActions/products.py
Normal file
140
UI_V2/dbActions/products.py
Normal file
@@ -0,0 +1,140 @@
|
||||
import sqlite3
|
||||
from typing import Optional
|
||||
|
||||
class Products:
|
||||
def __init__(self, db_path="instance/app_database.db"):
|
||||
self.db_path = db_path
|
||||
self._create_products_table()
|
||||
|
||||
def _create_products_table(self):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS products (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
name TEXT NOT NULL,
|
||||
description TEXT NOT NULL,
|
||||
details TEXT NOT NULL,
|
||||
price REAL,
|
||||
discount REAL,
|
||||
quantity REAL,
|
||||
aviability TEXT NOT NULL,
|
||||
category_id INTEGER,
|
||||
image TEXT NOT NULL,
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
|
||||
""")
|
||||
conn.commit()
|
||||
|
||||
def add(self, product) -> bool:
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
INSERT INTO products (name, description, details, price, discount, quantity, aviability, category_id, image)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""", (product['name'], product['description'],product['details'],product['price'], product['discount'], product['quantity'], product['aviability'], product['category_id'], product['image']))
|
||||
conn.commit()
|
||||
return True
|
||||
except sqlite3.IntegrityError:
|
||||
return False
|
||||
|
||||
def get(self, id):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT * FROM products
|
||||
WHERE id = ?
|
||||
""", (id,))
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return {
|
||||
"id": row[0],
|
||||
"name": row[1],
|
||||
"description": row[2],
|
||||
"details":row[3],
|
||||
"price": row[4],
|
||||
"discount": row[5],
|
||||
"quantity": row[6],
|
||||
"aviability": row[7],
|
||||
"category_id": row[8],
|
||||
"image":row[9],
|
||||
"created_at": row[10]
|
||||
}
|
||||
else:
|
||||
return None
|
||||
|
||||
def get_all(self):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT * FROM products
|
||||
""")
|
||||
rows = cursor.fetchall()
|
||||
if rows:
|
||||
products = []
|
||||
for row in rows:
|
||||
product={
|
||||
"id": row[0],
|
||||
"name": row[1],
|
||||
"description": row[2],
|
||||
"details":row[3],
|
||||
"price": row[4],
|
||||
"discount": row[5],
|
||||
"quantity": row[6],
|
||||
"aviability": row[7],
|
||||
"category_id": row[8],
|
||||
"image":row[9],
|
||||
"created_at": row[10]
|
||||
}
|
||||
products.append(product)
|
||||
return products
|
||||
else:
|
||||
return []
|
||||
|
||||
def get_all_by_category(self, category_id):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT * FROM products
|
||||
WHERE category_id = ?
|
||||
""", (category_id,))
|
||||
rows = cursor.fetchall()
|
||||
if rows:
|
||||
products = []
|
||||
for row in rows:
|
||||
product={
|
||||
"id": row[0],
|
||||
"name": row[1],
|
||||
"description": row[2],
|
||||
"details":row[3],
|
||||
"price": row[4],
|
||||
"discount": row[5],
|
||||
"quantity": row[6],
|
||||
"aviability": row[7],
|
||||
"category_id": row[8],
|
||||
"image": row[9],
|
||||
"created_at": row[10],
|
||||
}
|
||||
products.append(product)
|
||||
return products
|
||||
else:
|
||||
return []
|
||||
|
||||
def update(self, product, id):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
UPDATE products SET name = ?, description =?, details=?, price=?, discount=?, quantity=?, aviability=?, category_id=?, image=?
|
||||
WHERE id = ?
|
||||
''', (product['name'], product['description'], product['details'], product['price'], product['discount'], product['quantity'], product['aviability'], product['category_id'], product['image'] , id))
|
||||
conn.commit()
|
||||
|
||||
def delete(self, id):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
DELETE FROM products WHERE id=?;
|
||||
''', (id,))
|
||||
conn.commit()
|
||||
206
UI_V2/dbActions/users.py
Normal file
206
UI_V2/dbActions/users.py
Normal file
@@ -0,0 +1,206 @@
|
||||
import sqlite3
|
||||
from typing import Optional
|
||||
import hashlib
|
||||
|
||||
class Users:
|
||||
def __init__(self, db_path="instance/app_database.db"):
|
||||
self.db_path = db_path
|
||||
self._create_users_table()
|
||||
|
||||
def _create_users_table(self):
|
||||
"""Create the users 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 users (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
email TEXT NOT NULL UNIQUE,
|
||||
password TEXT,
|
||||
token TEXT,
|
||||
name TEXT,
|
||||
phone TEXT,
|
||||
address TEXT,
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
status TEXT NOT NULL DEFAULT 'active',
|
||||
role TEXT DEFAULT 'client'
|
||||
);
|
||||
|
||||
""")
|
||||
conn.commit()
|
||||
|
||||
def hash_password(self, password: str) -> bytes:
|
||||
return hashlib.md5(password.encode('utf-8')).hexdigest()
|
||||
|
||||
def add_user(self, email, passwd, role):
|
||||
"""Register a new user."""
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
INSERT INTO users (email, password, role)
|
||||
VALUES (?, ?, ?)
|
||||
""", (email, passwd, role))
|
||||
conn.commit()
|
||||
return True
|
||||
except sqlite3.IntegrityError:
|
||||
return False # Username already exist
|
||||
|
||||
def invite_user(self, email, name, phone, address, role='invited'):
|
||||
#try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
INSERT INTO users (email, name, phone, address, role)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
""", (email, name, phone, address, role))
|
||||
conn.commit()
|
||||
return cursor.lastrowid
|
||||
#except sqlite3.IntegrityError:
|
||||
# return None
|
||||
|
||||
def register_user(self, email: str, password: str) -> bool:
|
||||
"""Register a new user."""
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
INSERT INTO users (email, password)
|
||||
VALUES (?, ?)
|
||||
""", (email, password))
|
||||
conn.commit()
|
||||
return True
|
||||
except sqlite3.IntegrityError:
|
||||
return False # Username already exist
|
||||
|
||||
def authenticate_user(self, email: str, password: str) -> bool:
|
||||
"""Authenticate a user."""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT * FROM users
|
||||
WHERE email = ? AND password = ?
|
||||
""", (email, password))
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return {
|
||||
"id": row[0],
|
||||
"email": row[1],
|
||||
"token": row[3],
|
||||
"name":row[4],
|
||||
"phone": row[5],
|
||||
"address": row[6],
|
||||
"created_at": row[7],
|
||||
"status": row[8],
|
||||
"role":row[9]
|
||||
}
|
||||
else:
|
||||
return None
|
||||
|
||||
def get_user(self, email: str) -> Optional[dict]:
|
||||
"""Retrieve user details by username."""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT * FROM users
|
||||
WHERE email = ?
|
||||
""", (email,))
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return {
|
||||
"id": row[0],
|
||||
"email": row[1],
|
||||
"token": row[3],
|
||||
"name":row[4],
|
||||
"phone": row[5],
|
||||
"address": row[6],
|
||||
"created_at": row[7],
|
||||
"status": row[8],
|
||||
"role":row[9]
|
||||
}
|
||||
return None
|
||||
|
||||
def get(self, id: int) -> Optional[dict]:
|
||||
"""Retrieve user details by username."""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT * FROM users
|
||||
WHERE id = ?
|
||||
""", (id,))
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return {
|
||||
"id": row[0],
|
||||
"email": row[1],
|
||||
"token": row[3],
|
||||
"name":row[4],
|
||||
"phone": row[5],
|
||||
"address": row[6],
|
||||
"created_at": row[7],
|
||||
"status": row[8],
|
||||
"role":row[9]
|
||||
}
|
||||
return None
|
||||
|
||||
def get_all(self) -> Optional[dict]:
|
||||
"""Retrieve user details by username."""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT * FROM users
|
||||
""")
|
||||
rows = cursor.fetchall()
|
||||
if rows:
|
||||
buffer = []
|
||||
for row in rows:
|
||||
buffer.append({
|
||||
"id": row[0],
|
||||
"email": row[1],
|
||||
"token": row[3],
|
||||
"name":row[4],
|
||||
"phone": row[5],
|
||||
"address": row[6],
|
||||
"created_at": row[7],
|
||||
"status": row[8],
|
||||
"role":row[9]
|
||||
})
|
||||
return buffer
|
||||
return []
|
||||
|
||||
def update_password(self, email, passwd):
|
||||
'''Update user password'''
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
UPDATE users SET password = ?
|
||||
WHERE email = ?
|
||||
''', (passwd, email))
|
||||
conn.commit()
|
||||
|
||||
def update_token(self, id, token):
|
||||
'''Update user token'''
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
UPDATE users SET token = ?
|
||||
WHERE id = ?
|
||||
''', (token, id))
|
||||
conn.commit()
|
||||
|
||||
def update_user_data(self, name, phone, address, id):
|
||||
'''Update user data'''
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
UPDATE users SET name = ?, phone = ?, address = ?
|
||||
WHERE id = ?
|
||||
''', (name, phone, address, id))
|
||||
conn.commit()
|
||||
|
||||
def delete(self, id):
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
DELETE FROM users WHERE id=?;
|
||||
''', (id,))
|
||||
conn.commit()
|
||||
Reference in New Issue
Block a user