ske087/quality_app-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix PyMySQL DictCursor usage in warehouse functions

Browse Source 
Fixed incorrect pymysql.cursors.DictCursor instantiation in 4 functions:
- get_cp_inventory_list()
- search_cp_code()
- search_by_box_number()
- get_cp_details()

PyMySQL connection pool doesn't support cursor_factory parameter.
Now manually converting tuples to dicts using cursor.description instead.
This commit is contained in:
Quality App Developer
2026-01-30 15:20:49 +02:00
parent f97b9692b8
commit 39a3a0084c
1 changed files with 16 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
app/modules/warehouse/warehouse.py
View File

@@ -449,7 +449,7 @@ def get_cp_inventory_list(limit=100, offset=0):
""" """
try: try:
conn = get_db() conn = get_db()
cursor = conn.cursor(pymysql.cursors.DictCursor) cursor = conn.cursor()
# Get all CP codes with their box and location info (latest entries first) # Get all CP codes with their box and location info (latest entries first)
cursor.execute(""" cursor.execute("""
@@ -474,7 +474,9 @@ def get_cp_inventory_list(limit=100, offset=0):
LIMIT %s OFFSET %s LIMIT %s OFFSET %s
""", (limit, offset)) """, (limit, offset))
results = cursor.fetchall() # Convert tuples to dicts using cursor description
columns = [col[0] for col in cursor.description]
results = [{columns[i]: row[i] for i in range(len(columns))} for row in cursor.fetchall()]
cursor.close() cursor.close()
return results if results else [] return results if results else []
@@ -495,7 +497,7 @@ def search_cp_code(cp_code_search):
""" """
try: try:
conn = get_db() conn = get_db()
cursor = conn.cursor(pymysql.cursors.DictCursor) cursor = conn.cursor()
# Remove hyphen and get base CP code if full CP provided # Remove hyphen and get base CP code if full CP provided
search_term = cp_code_search.replace('-', '').strip().upper() search_term = cp_code_search.replace('-', '').strip().upper()
@@ -523,7 +525,9 @@ def search_cp_code(cp_code_search):
ORDER BY MAX(s.created_at) DESC ORDER BY MAX(s.created_at) DESC
""", (f"{search_term}%",)) """, (f"{search_term}%",))
results = cursor.fetchall() # Convert tuples to dicts using cursor description
columns = [col[0] for col in cursor.description]
results = [{columns[i]: row[i] for i in range(len(columns))} for row in cursor.fetchall()]
cursor.close() cursor.close()
return results if results else [] return results if results else []
@@ -544,7 +548,7 @@ def search_by_box_number(box_number_search):
""" """
try: try:
conn = get_db() conn = get_db()
cursor = conn.cursor(pymysql.cursors.DictCursor) cursor = conn.cursor()
box_search = box_number_search.strip().upper() box_search = box_number_search.strip().upper()
@@ -571,7 +575,9 @@ def search_by_box_number(box_number_search):
LIMIT 500 LIMIT 500
""", (f"%{box_search}%",)) """, (f"%{box_search}%",))
results = cursor.fetchall() # Convert tuples to dicts using cursor description
columns = [col[0] for col in cursor.description]
results = [{columns[i]: row[i] for i in range(len(columns))} for row in cursor.fetchall()]
cursor.close() cursor.close()
return results if results else [] return results if results else []
@@ -593,7 +599,7 @@ def get_cp_details(cp_code):
""" """
try: try:
conn = get_db() conn = get_db()
cursor = conn.cursor(pymysql.cursors.DictCursor) cursor = conn.cursor()
# Search for all entries with this CP base # Search for all entries with this CP base
cursor.execute(""" cursor.execute("""
@@ -622,7 +628,9 @@ def get_cp_details(cp_code):
LIMIT 1000 LIMIT 1000
""", (cp_code.upper(),)) """, (cp_code.upper(),))
results = cursor.fetchall() # Convert tuples to dicts using cursor description
columns = [col[0] for col in cursor.description]
results = [{columns[i]: row[i] for i in range(len(columns))} for row in cursor.fetchall()]
cursor.close() cursor.close()
return results if results else [] return results if results else []