ske087/quality_recticel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

updated database deployment and views

Browse Source
This commit is contained in:
Quality System Admin
2025-10-12 00:22:45 +03:00
parent a84c881e71
commit aaf6f2b32f
6 changed files with 563 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
py_app/app/warehouse.py
View File

@@ -89,7 +89,7 @@ def delete_locations_by_ids(ids_str):
cursor = conn.cursor()
deleted = 0
for id in ids:
cursor.execute("DELETE FROM warehouse_locations WHERE id = ?", (id,))
cursor.execute("DELETE FROM warehouse_locations WHERE id = %s", (id,))
if cursor.rowcount:
deleted += 1
conn.commit()
@@ -226,20 +226,20 @@ def update_location(location_id, location_code, size, description):
cursor = conn.cursor()
# Check if location exists
cursor.execute("SELECT id FROM warehouse_locations WHERE id = ?", (location_id,))
cursor.execute("SELECT id FROM warehouse_locations WHERE id = %s", (location_id,))
if not cursor.fetchone():
conn.close()
return {"success": False, "error": "Location not found"}
# Check if location code already exists for different location
cursor.execute("SELECT id FROM warehouse_locations WHERE location_code = ? AND id != ?", (location_code, location_id))
cursor.execute("SELECT id FROM warehouse_locations WHERE location_code = %s AND id != %s", (location_code, location_id))
if cursor.fetchone():
conn.close()
return {"success": False, "error": "Location code already exists"}
# Update location
cursor.execute(
"UPDATE warehouse_locations SET location_code = ?, size = ?, description = ? WHERE id = ?",
"UPDATE warehouse_locations SET location_code = %s, size = %s, description = %s WHERE id = %s",
(location_code, size if size else None, description, location_id)
)
conn.commit()
@@ -250,6 +250,8 @@ def update_location(location_id, location_code, size, description):
except Exception as e:
print(f"Error updating location: {e}")
return {"success": False, "error": str(e)}
print(f"Error updating location: {e}")
return {"success": False, "error": str(e)}
def delete_location_by_id(location_id):
"""Delete a warehouse location by ID"""
@@ -258,14 +260,14 @@ def delete_location_by_id(location_id):
cursor = conn.cursor()
# Check if location exists
cursor.execute("SELECT location_code FROM warehouse_locations WHERE id = ?", (location_id,))
cursor.execute("SELECT location_code FROM warehouse_locations WHERE id = %s", (location_id,))
location = cursor.fetchone()
if not location:
conn.close()
return {"success": False, "error": "Location not found"}
# Delete location
cursor.execute("DELETE FROM warehouse_locations WHERE id = ?", (location_id,))
cursor.execute("DELETE FROM warehouse_locations WHERE id = %s", (location_id,))
conn.commit()
conn.close()