127 lines
4.4 KiB
Python
127 lines
4.4 KiB
Python
import sqlite3
|
|
from flask import current_app as app
|
|
import time
|
|
|
|
def get_db_connection():
|
|
conn = sqlite3.connect('./static/db/db.sqlite')
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
def dict_from_row(row):
|
|
return {key: row[key] for key in row.keys()}
|
|
|
|
def execute_with_retry(conn, query, params=(), retries=5, delay=0.1):
|
|
for attempt in range(retries):
|
|
try:
|
|
conn.execute(query, params)
|
|
conn.commit()
|
|
return
|
|
except sqlite3.OperationalError as e:
|
|
if "database is locked" in str(e):
|
|
if attempt < retries - 1:
|
|
time.sleep(delay)
|
|
else:
|
|
raise
|
|
else:
|
|
raise
|
|
|
|
def fetch_users(role_id):
|
|
conn = get_db_connection()
|
|
users = conn.execute('SELECT * FROM Zamestnanci WHERE Role_ID >= ?', (role_id,)).fetchall()
|
|
users = [dict_from_row(user) for user in users]
|
|
conn.close()
|
|
app.logger.debug(f"Fetched users: {users}")
|
|
return users
|
|
|
|
def fetch_orders():
|
|
conn = get_db_connection()
|
|
orders = conn.execute('SELECT * FROM Objednavky').fetchall()
|
|
orders = [dict_from_row(order) for order in orders]
|
|
conn.close()
|
|
app.logger.debug(f"Fetched orders: {orders}")
|
|
return orders
|
|
|
|
def fetch_roles():
|
|
conn = get_db_connection()
|
|
roles = conn.execute('SELECT * FROM Role').fetchall()
|
|
roles = [dict_from_row(role) for role in roles]
|
|
conn.close()
|
|
app.logger.debug(f"Fetched roles: {roles}")
|
|
return roles
|
|
|
|
def fetch_repairs():
|
|
conn = get_db_connection()
|
|
repairs = conn.execute('''
|
|
SELECT Opravy.ID_Opravy, Zamestnanci.Jmeno, Zamestnanci.Prijmeni, Opravy.Nazev, Opravy.Popis
|
|
FROM Opravy
|
|
JOIN Zamestnanci ON Opravy.ID_Zamestnance = Zamestnanci.ID_Uzivatele
|
|
''').fetchall()
|
|
repairs = [dict_from_row(repair) for repair in repairs]
|
|
|
|
for repair in repairs:
|
|
products = conn.execute('''
|
|
SELECT Produkty.Nazev, Pouzite_Produkty.Pocet_Produktu
|
|
FROM Pouzite_Produkty
|
|
JOIN Produkty ON Pouzite_Produkty.ID_Produktu = Produkty.ID_Produktu
|
|
WHERE Pouzite_Produkty.ID_Opravy = ?
|
|
''', (repair['ID_Opravy'],)).fetchall()
|
|
repair['products'] = [dict_from_row(product) for product in products]
|
|
|
|
conn.close()
|
|
app.logger.debug(f"Fetched repairs: {repairs}")
|
|
return repairs
|
|
|
|
def fetch_employees():
|
|
conn = get_db_connection()
|
|
employees = conn.execute('SELECT ID_Uzivatele, Jmeno, Prijmeni FROM Zamestnanci').fetchall()
|
|
employees = [dict_from_row(employee) for employee in employees]
|
|
conn.close()
|
|
app.logger.debug(f"Fetched employees: {employees}")
|
|
return employees
|
|
|
|
def fetch_products():
|
|
conn = get_db_connection()
|
|
products = conn.execute('SELECT * FROM Produkty').fetchall()
|
|
products = [dict_from_row(product) for product in products]
|
|
conn.close()
|
|
app.logger.debug(f"Fetched products: {products}")
|
|
return products
|
|
|
|
def update_product(product_id, nazev, popis, momentalni_zasoba, minimalni_zasoba):
|
|
conn = get_db_connection()
|
|
execute_with_retry(conn, '''
|
|
UPDATE Produkty
|
|
SET Nazev = ?, Popis = ?, Momentalni_Zasoba = ?, Minimalni_Zasoba = ?
|
|
WHERE ID_Produktu = ?
|
|
''', (nazev, popis, momentalni_zasoba, minimalni_zasoba, product_id))
|
|
conn.close()
|
|
|
|
def add_product_stock(product_id, quantity):
|
|
conn = get_db_connection()
|
|
execute_with_retry(conn, '''
|
|
UPDATE Produkty
|
|
SET Momentalni_Zasoba = Momentalni_Zasoba + ?
|
|
WHERE ID_Produktu = ?
|
|
''', (quantity, product_id))
|
|
conn.close()
|
|
|
|
def create_product(nazev, popis, momentalni_zasoba, minimalni_zasoba):
|
|
conn = get_db_connection()
|
|
execute_with_retry(conn, '''
|
|
INSERT INTO Produkty (Nazev, Popis, Momentalni_Zasoba, Minimalni_Zasoba)
|
|
VALUES (?, ?, ?, ?)
|
|
''', (nazev, popis, momentalni_zasoba, minimalni_zasoba))
|
|
conn.close()
|
|
|
|
def insert_used_products(repair_id, products):
|
|
conn = get_db_connection()
|
|
try:
|
|
for product in products:
|
|
product_id = product['id']
|
|
quantity = product['quantity']
|
|
app.logger.debug(f"Inserting product {product_id} with quantity {quantity} for repair {repair_id}")
|
|
if quantity and int(quantity) > 0:
|
|
execute_with_retry(conn, 'INSERT INTO Pouzite_Produkty (ID_Opravy, ID_Produktu, Pocet_Produktu) VALUES (?, ?, ?)',
|
|
(repair_id, product_id, quantity))
|
|
finally:
|
|
conn.close() |