diff --git a/app/general/general_utils/sql_service.py b/app/general/general_utils/sql_service.py index 5c5426d..1d3bc58 100644 --- a/app/general/general_utils/sql_service.py +++ b/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) - rows = cursor.fetchall() - - columns = [column[0] for column in cursor.description] - result = [dict(zip(columns, row)) for row in rows] + result = await conn.exec_driver_sql(sql) + result = [dict(row._mapping) for row in result] return [val for key, val in result[0].items()] -def list(connect_config, query_config, module_config, debug_data=[]): - conn, cursor, close = get_sql_connection(connect_config) +async def list(session: AsyncSessionDep, query_config, module_config, debug_data=[]): + 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) - rows = cursor.fetchall() - - # 将查询结果转换为字典列表 - columns = [column[0] for column in cursor.description] - result = [dict(zip(columns, row)) for row in rows] + result = await conn.exec_driver_sql(sql, tuple(values)) + result = [dict(row._mapping) for row in result] 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=[]): - conn, cursor, close = get_sql_connection(connect_config) +async def insert(session: AsyncSessionDep, query_config, module_config, debug_data=[]): + 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) - conn.commit() + await conn.exec_driver_sql(sql, tuple(values)) + await session.commit() - sql, values = build_query_sql({"page": 0, "size": 1, "filters": [{"field": "id", "operator": "=", "value": row_id}]}, module_config) - debug_data.append({"sql": sql, "values": values}) - cursor.execute(sql, values) - rows = cursor.fetchall() + result = await item(session, query_config={"id": row_id}, module_config=module_config, debug_data=debug_data) - columns = [column[0] for column in cursor.description] - result = [dict(zip(columns, row)) for row in rows] + if "error" in result: + 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=[]): - conn, cursor, close = get_sql_connection(connect_config) +async def batch_insert(session: AsyncSessionDep, query_config, module_config, debug_data=[]): + 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 = [] - for row in rows: - row_id = get_value(row, 'id', None) - if row_id is None: - row_id = get_id(cursor, 1)[0] - row['id'] = row_id - row_id_list.append(row_id) + rows_without_id = [row for row in rows if get_value(row, 'id', None) is None] + if rows_without_id: + new_id_list = await get_id(session, len(rows_without_id)) + for index, row in enumerate(rows_without_id): + row['id'] = new_id_list[index] 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) - debug_data.append({"sql": sql, "values": values}) - cursor.execute(sql, values) - rows = cursor.fetchall() + row_id_list = [row['id'] for row in rows] + query_config = {"all": True, "filters": [{"field": "id", "operator": "in", "value": row_id_list}]} - columns = [column[0] for column in cursor.description] - result = [dict(zip(columns, row)) for row in rows] - - create_convertor(module_config)['decode_list'](result) + result = await list(session, query_config=query_config, module_config=module_config, debug_data=debug_data) + result = get_value(result, 'list', []) 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=[]): - conn, cursor, close = get_sql_connection(connect_config) +async def update(session: AsyncSessionDep, query_config, module_config, debug_data=[]): + 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) - conn.commit() + await conn.exec_driver_sql(sql, tuple(values)) + await session.commit() - sql, values = build_query_sql({"page": 0, "size": 1, "filters": [{"field": "id", "operator": "=", "value": row_id}]}, module_config) - debug_data.append({"sql": sql, "values": values}) - cursor.execute(sql, values) - rows = cursor.fetchall() + result = await item(session, query_config={"id": row_id}, module_config=module_config, debug_data=debug_data) - columns = [column[0] for column in cursor.description] - result = [dict(zip(columns, row)) for row in rows] + if "error" in result: + 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=[]): - conn, cursor, close = get_sql_connection(connect_config) +async def batch_update(session: AsyncSessionDep, query_config, module_config, debug_data=[]): + 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 = [] - for row in rows: - row_id = get_value(row, 'id', None) - if row_id is None: - return { - "error": "row is missing field: id", - "row": row, - } - row_id_list.append(row_id) + rows_without_id = [row for row in rows if get_value(row, 'id', None) is None] + if rows_without_id: + return { + "error": "row is missing field: id", + "rows": rows_without_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) - debug_data.append({"sql": sql, "values": values}) - cursor.execute(sql, values) - rows = cursor.fetchall() + row_id_list = [row['id'] for row in rows] + query_config = {"all": True, "filters": [{"field": "id", "operator": "in", "value": row_id_list}]} - columns = [column[0] for column in cursor.description] - result = [dict(zip(columns, row)) for row in rows] - - create_convertor(module_config)['decode_list'](result) + result = await list(session, query_config=query_config, module_config=module_config, debug_data=debug_data) + result = get_value(result, 'list', []) 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=[]): - conn, cursor, close = get_sql_connection(connect_config) +async def delete(session: AsyncSessionDep, query_config, module_config, debug_data=[]): + 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) - conn.commit() - - deleted_rows = cursor.rowcount + result = await conn.exec_driver_sql(sql, tuple(values)) + await session.commit() + deleted_rows = result.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():