feat: 迁移异步的sql_service

app/general/general_utils/sql_service.py

@@ -1,8 +1,11 @@
+import traceback
+
 from app.general.general_utils.build_delete_sql import build_delete_sql
 from app.general.general_utils.build_insert_sql import build_insert_sql
 from app.general.general_utils.build_query_sql import build_query_sql
 from app.general.general_utils.build_update_sql import build_update_sql
 from app.general.general_utils.sql_utils import get_value, create_convertor
+from app.utils.db_utils import AsyncSessionDep
 
 
 def get_default_orders(query_config, module_config):
@@ -19,21 +22,19 @@ def get_default_orders(query_config, module_config):
   return {"field": "createdAt", "desc": True}
 
 
-def get_id(cursor, len):
+async def get_id(session: AsyncSessionDep, len: int | None = 1):
+  conn = await session.connection()
   if len is None:
     len = 1
   sql = f"select {','.join([f'uuid() as _{idx}' for idx in range(len)])}"
-  cursor.execute(sql)
+  result = await conn.exec_driver_sql(sql)
-  rows = cursor.fetchall()
+  result = [dict(row._mapping) for row in result]
-
-  columns = [column[0] for column in cursor.description]
-  result = [dict(zip(columns, row)) for row in rows]
 
   return [val for key, val in result[0].items()]
 
 
-def list(connect_config, query_config, module_config, debug_data=[]):
+async def list(session: AsyncSessionDep, query_config, module_config, debug_data=[]):
-  conn, cursor, close = get_sql_connection(connect_config)
+  conn = await session.connection()
 
   n_page = get_value(query_config, 'page', 0)
   n_size = get_value(query_config, 'size', 5)
@@ -55,12 +56,8 @@ def list(connect_config, query_config, module_config, debug_data=[]):
   try:
 
     debug_data.append({"sql": sql, "values": values})
-    cursor.execute(sql, values)
+    result = await conn.exec_driver_sql(sql, tuple(values))
-    rows = cursor.fetchall()
+    result = [dict(row._mapping) for row in result]
-
-    # 将查询结果转换为字典列表
-    columns = [column[0] for column in cursor.description]
-    result = [dict(zip(columns, row)) for row in rows]
 
     create_convertor(module_config)['decode_list'](result)
 
@@ -76,15 +73,15 @@ def list(connect_config, query_config, module_config, debug_data=[]):
         "hasNext": has_next,
         "list": result,
       }
-  except mysql.connector.Error as err:
+  except Exception as err:
+    print(err)
+    traceback.print_exc()
     return {
       "error": f"Error: {err}",
     }
-  finally:
-    close()
 
 
-def item(connect_config, query_config, module_config, debug_data=[]):
+async def item(session: AsyncSessionDep, query_config, module_config, debug_data=[]):
   target_query_config = {
     "offset": 0,
     "size": 1,
@@ -98,7 +95,7 @@ def item(connect_config, query_config, module_config, debug_data=[]):
       "operator": "="
     })
 
-  result = list(connect_config=connect_config, query_config=target_query_config, module_config=module_config, debug_data=debug_data)
+  result = await list(session=session, query_config=target_query_config, module_config=module_config, debug_data=debug_data)
 
   if "error" in result:
     return result
@@ -106,8 +103,8 @@ def item(connect_config, query_config, module_config, debug_data=[]):
   return {"result": None if "list" not in result or len(result['list']) == 0 else result['list'][0]}
 
 
-def insert(connect_config, query_config, module_config, debug_data=[]):
+async def insert(session: AsyncSessionDep, query_config, module_config, debug_data=[]):
-  conn, cursor, close = get_sql_connection(connect_config)
+  conn = await session.connection()
 
   row = get_value(query_config, 'row', None)
   if row is None:
@@ -119,45 +116,42 @@ def insert(connect_config, query_config, module_config, debug_data=[]):
   row_id = get_value(row, 'id', None)
 
   if row_id is None:
-    row_id = get_id(cursor, 1)[0]
+    row_id = (await get_id(session, 1))[0]
     row['id'] = row_id
 
   try:
     sql, values = build_insert_sql(module_config, row)
 
     debug_data.append({"sql": sql, "values": values})
-    cursor.execute(sql, values)
+    await conn.exec_driver_sql(sql, tuple(values))
-    conn.commit()
+    await session.commit()
 
-    sql, values = build_query_sql({"page": 0, "size": 1, "filters": [{"field": "id", "operator": "=", "value": row_id}]}, module_config)
+    result = await item(session, query_config={"id": row_id}, module_config=module_config, debug_data=debug_data)
-    debug_data.append({"sql": sql, "values": values})
-    cursor.execute(sql, values)
-    rows = cursor.fetchall()
 
-    columns = [column[0] for column in cursor.description]
+    if "error" in result:
-    result = [dict(zip(columns, row)) for row in rows]
+      return result
 
-    create_convertor(module_config)['decode_list'](result)
+    item_dict = get_value(result, 'result', None)
 
-    if len(result) > 0:
+    if item_dict is not None:
       return {
-        "result": result[0]
+        "result": item_dict
       }
     else:
       return {
         "error": "insert failed, query result is empty",
       }
 
-  except mysql.connector.Error as err:
+  except Exception as err:
+    print(err)
+    traceback.print_exc()
     return {
       "error": f"Error: {err}",
     }
-  finally:
-    close()
 
 
-def batch_insert(connect_config, query_config, module_config, debug_data=[]):
+async def batch_insert(session: AsyncSessionDep, query_config, module_config, debug_data=[]):
-  conn, cursor, close = get_sql_connection(connect_config)
+  conn = await session.connection()
 
   rows = get_value(query_config, 'rows', None)
   if rows is None or len(rows) == 0:
@@ -166,31 +160,25 @@ def batch_insert(connect_config, query_config, module_config, debug_data=[]):
     }
   create_convertor(module_config)['encode_list'](rows)
 
-  row_id_list = []
+  rows_without_id = [row for row in rows if get_value(row, 'id', None) is None]
-  for row in rows:
+  if rows_without_id:
-    row_id = get_value(row, 'id', None)
+    new_id_list = await get_id(session, len(rows_without_id))
-    if row_id is None:
+    for index, row in enumerate(rows_without_id):
-      row_id = get_id(cursor, 1)[0]
+      row['id'] = new_id_list[index]
-      row['id'] = row_id
-    row_id_list.append(row_id)
 
   try:
     for row in rows:
       sql, values = build_insert_sql(module_config, row)
       debug_data.append({"sql": sql, "values": values})
-      cursor.execute(sql, values)
+      await conn.exec_driver_sql(sql, tuple(values))
 
-    conn.commit()
+    await session.commit()
 
-    sql, values = build_query_sql({"all": True, "filters": [{"field": "id", "operator": "in", "value": row_id_list}]}, module_config)
+    row_id_list = [row['id'] for row in rows]
-    debug_data.append({"sql": sql, "values": values})
+    query_config = {"all": True, "filters": [{"field": "id", "operator": "in", "value": row_id_list}]}
-    cursor.execute(sql, values)
-    rows = cursor.fetchall()
 
-    columns = [column[0] for column in cursor.description]
+    result = await list(session, query_config=query_config, module_config=module_config, debug_data=debug_data)
-    result = [dict(zip(columns, row)) for row in rows]
+    result = get_value(result, 'list', [])
-
-    create_convertor(module_config)['decode_list'](result)
 
     if len(result) > 0:
       return {
@@ -201,16 +189,16 @@ def batch_insert(connect_config, query_config, module_config, debug_data=[]):
         "error": "insert failed, query result is empty",
       }
 
-  except mysql.connector.Error as err:
+  except Exception as err:
+    print(err)
+    traceback.print_exc()
     return {
       "error": f"Error: {err}",
     }
-  finally:
-    close()
 
 
-def update(connect_config, query_config, module_config, debug_data=[]):
+async def update(session: AsyncSessionDep, query_config, module_config, debug_data=[]):
-  conn, cursor, close = get_sql_connection(connect_config)
+  conn = await session.connection()
 
   row = get_value(query_config, 'row', None)
   update_fields = get_value(query_config, 'updateFields', None)
@@ -233,38 +221,35 @@ def update(connect_config, query_config, module_config, debug_data=[]):
     sql, values = build_update_sql(module_config, row, update_fields)
 
     debug_data.append({"sql": sql, "values": values})
-    cursor.execute(sql, values)
+    await conn.exec_driver_sql(sql, tuple(values))
-    conn.commit()
+    await session.commit()
 
-    sql, values = build_query_sql({"page": 0, "size": 1, "filters": [{"field": "id", "operator": "=", "value": row_id}]}, module_config)
+    result = await item(session, query_config={"id": row_id}, module_config=module_config, debug_data=debug_data)
-    debug_data.append({"sql": sql, "values": values})
-    cursor.execute(sql, values)
-    rows = cursor.fetchall()
 
-    columns = [column[0] for column in cursor.description]
+    if "error" in result:
-    result = [dict(zip(columns, row)) for row in rows]
+      return result
 
-    create_convertor(module_config)['decode_list'](result)
+    item_dict = get_value(result, 'result', None)
 
-    if len(result) > 0:
+    if item_dict is not None:
       return {
-        "result": result[0]
+        "result": item_dict
       }
     else:
       return {
         "error": "update failed, query result is empty",
       }
 
-  except mysql.connector.Error as err:
+  except Exception as err:
+    print(err)
+    traceback.print_exc()
     return {
       "error": f"Error: {err}",
     }
-  finally:
-    close()
 
 
-def batch_update(connect_config, query_config, module_config, debug_data=[]):
+async def batch_update(session: AsyncSessionDep, query_config, module_config, debug_data=[]):
-  conn, cursor, close = get_sql_connection(connect_config)
+  conn = await session.connection()
 
   rows = get_value(query_config, 'rows', None)
   update_fields = get_value(query_config, 'updateFields', None)
@@ -275,33 +260,26 @@ def batch_update(connect_config, query_config, module_config, debug_data=[]):
     }
   create_convertor(module_config)['encode_list'](rows)
 
-  row_id_list = []
+  rows_without_id = [row for row in rows if get_value(row, 'id', None) is None]
-  for row in rows:
+  if rows_without_id:
-    row_id = get_value(row, 'id', None)
-    if row_id is None:
     return {
       "error": "row is missing field: id",
-        "row": row,
+      "rows": rows_without_id,
     }
-    row_id_list.append(row_id)
 
   try:
     for row in rows:
       sql, values = build_update_sql(module_config, row, update_fields)
       debug_data.append({"sql": sql, "values": values})
-      cursor.execute(sql, values)
+      await conn.exec_driver_sql(sql, tuple(values))
 
-    conn.commit()
+    await session.commit()
 
-    sql, values = build_query_sql({"all": True, "filters": [{"field": "id", "operator": "in", "value": row_id_list}]}, module_config)
+    row_id_list = [row['id'] for row in rows]
-    debug_data.append({"sql": sql, "values": values})
+    query_config = {"all": True, "filters": [{"field": "id", "operator": "in", "value": row_id_list}]}
-    cursor.execute(sql, values)
-    rows = cursor.fetchall()
 
-    columns = [column[0] for column in cursor.description]
+    result = await list(session, query_config=query_config, module_config=module_config, debug_data=debug_data)
-    result = [dict(zip(columns, row)) for row in rows]
+    result = get_value(result, 'list', [])
-
-    create_convertor(module_config)['decode_list'](result)
 
     if len(result) > 0:
       return {
@@ -312,16 +290,17 @@ def batch_update(connect_config, query_config, module_config, debug_data=[]):
         "error": "update failed, query result is empty",
       }
 
-  except mysql.connector.Error as err:
+  except Exception as err:
+    print(err)
+    traceback.print_exc()
     return {
       "error": f"Error: {err}",
     }
-  finally:
-    close()
 
 
-def delete(connect_config, query_config, module_config, debug_data=[]):
+async def delete(session: AsyncSessionDep, query_config, module_config, debug_data=[]):
-  conn, cursor, close = get_sql_connection(connect_config)
+  conn = await session.connection()
+
   id = get_value(query_config, 'id', None)
   if id is None:
     return {
@@ -331,21 +310,20 @@ def delete(connect_config, query_config, module_config, debug_data=[]):
   try:
     sql, values = build_delete_sql(module_config, id)
     debug_data.append({"sql": sql, "values": values})
-    cursor.execute(sql, values)
+    result = await conn.exec_driver_sql(sql, tuple(values))
-    conn.commit()
+    await session.commit()
-
+    deleted_rows = result.rowcount
-    deleted_rows = cursor.rowcount
 
     if deleted_rows >= 1:
       return {"deletedRows": deleted_rows}
     else:
       return {"error": f"delete failed, delete rows is {deleted_rows}", }
-  except mysql.connector.Error as err:
+  except Exception as err:
+    print(err)
+    traceback.print_exc()
     return {
       "error": f"Error: {err}",
     }
-  finally:
-    close()
 
 
 class SqlService():