ai-admin-server/app/general/general_utils/sql_service.py

import traceback

from app.config.env import env
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):
  orders = get_value(query_config, 'orders', None)
  if orders is not None:
    return orders
  module_config_default_orders = get_value(
    get_value(module_config, 'default', {}),
    'orders',
    None
  )
  if module_config_default_orders is not None:
    return module_config_default_orders
  return {"field": "createdAt", "desc": True}


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)])}"
  result = await conn.exec_driver_sql(sql)
  result = [dict(row._mapping) for row in result]

  return [val for key, val in result[0].items()]


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)
  n_only_count = get_value(query_config, 'onlyCount', False)

  offset = n_page * n_size
  # 多查一条数据，方便判断是否有下一页数据
  size = n_size + 1

  target_query_config = {
    **query_config,
    "offset": offset,
    "size": size,
    "orders": get_default_orders(query_config, module_config)
  }

  sql, values = build_query_sql(target_query_config, module_config)

  try:

    debug_data.append({"sql": sql, "values": values})
    result = await conn.exec_driver_sql(sql, tuple(values))
    result = [dict(row._mapping) for row in result]

    create_convertor(module_config)['decode_list'](result)

    if n_only_count:
      return {
        "total": result[0]['total']
      }
    else:
      has_next = False if get_value(query_config, 'all', False) else len(result) == n_size + 1
      if has_next:
        result.pop()
      return {
        "hasNext": has_next,
        "list": result,
      }
  except Exception as err:
    print(err)
    traceback.print_exc()
    return {
      "error": f"Error: {err}",
    }


async def item(session: AsyncSessionDep, query_config, module_config, debug_data=[]):
  target_query_config = {
    "offset": 0,
    "size": 1,
    "filters": [],
    "orders": {"field": "createdAt", "desc": True},
  }
  for humpName, value in query_config.items():
    target_query_config['filters'].append({
      "field": humpName,
      "value": value,
      "operator": "="
    })

  result = await list(session=session, query_config=target_query_config, module_config=module_config, debug_data=debug_data)

  if "error" in result:
    return result

  return {"result": None if "list" not in result or len(result['list']) == 0 else result['list'][0]}


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:
    return {
      "error": "row parameter is missing",
    }
  create_convertor(module_config)['encode_list']([row])

  row_id = get_value(row, 'id', None)

  if row_id is None:
    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})
    await conn.exec_driver_sql(sql, tuple(values))
    await session.commit()

    result = await item(session, query_config={"id": row_id}, module_config=module_config, debug_data=debug_data)

    if "error" in result:
      return result

    item_dict = get_value(result, 'result', None)

    if item_dict is not None:
      return {
        "result": item_dict
      }
    else:
      return {
        "error": "insert failed, query result is empty",
      }

  except Exception as err:
    print(err)
    traceback.print_exc()
    return {
      "error": f"Error: {err}",
    }


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:
    return {
      "error": "rows parameter is missing",
    }
  create_convertor(module_config)['encode_list'](rows)

  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})
      await conn.exec_driver_sql(sql, tuple(values))

    await session.commit()

    row_id_list = [row['id'] for row in rows]
    query_config = {"all": True, "filters": [{"field": "id", "operator": "in", "value": row_id_list}]}

    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 {
        "result": result
      }
    else:
      return {
        "error": "insert failed, query result is empty",
      }

  except Exception as err:
    print(err)
    traceback.print_exc()
    return {
      "error": f"Error: {err}",
    }


async def update(session: AsyncSessionDep, query_config, module_config, debug_data=[]):
  conn = await session.connection()

  row = get_value(query_config, 'row', None)
  update_by_fields = get_value(query_config, 'updateByFields', env.default_update_by_fields)

  if row is None:
    return {
      "error": "row parameter is missing",
    }

  row_id = get_value(row, 'id', None)

  if row_id is None:
    return {
      "error": "row is missing field: id",
    }

  create_convertor(module_config)['encode_list']([row])

  try:
    sql, values = build_update_sql(module_config, row, row.keys() if update_by_fields else None)

    debug_data.append({"sql": sql, "values": values})
    await conn.exec_driver_sql(sql, tuple(values))
    await session.commit()

    result = await item(session, query_config={"id": row_id}, module_config=module_config, debug_data=debug_data)

    if "error" in result:
      return result

    item_dict = get_value(result, 'result', None)

    if item_dict is not None:
      return {
        "result": item_dict
      }
    else:
      return {
        "error": "update failed, query result is empty",
      }

  except Exception as err:
    print(err)
    traceback.print_exc()
    return {
      "error": f"Error: {err}",
    }


async def batch_update(session: AsyncSessionDep, query_config, module_config, debug_data=[]):
  conn = await session.connection()

  rows = get_value(query_config, 'rows', None)
  update_by_fields = get_value(query_config, 'updateByFields', env.default_update_by_fields)

  if rows is None or len(rows) == 0:
    return {
      "error": "rows parameter is missing",
    }
  create_convertor(module_config)['encode_list'](rows)

  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, row.keys() if update_by_fields else None)
      debug_data.append({"sql": sql, "values": values})
      await conn.exec_driver_sql(sql, tuple(values))

    await session.commit()

    row_id_list = [row['id'] for row in rows]
    query_config = {"all": True, "filters": [{"field": "id", "operator": "in", "value": row_id_list}]}

    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 {
        "result": result
      }
    else:
      return {
        "error": "update failed, query result is empty",
      }

  except Exception as err:
    print(err)
    traceback.print_exc()
    return {
      "error": f"Error: {err}",
    }


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 {
      "error": "id parameter is missing",
    }

  try:
    sql, values = build_delete_sql(module_config, id)
    debug_data.append({"sql": sql, "values": values})
    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 Exception as err:
    print(err)
    traceback.print_exc()
    return {
      "error": f"Error: {err}",
    }


class SqlService():
  def __init__(self):
    self.list = list
    self.item = item
    self.insert = insert
    self.update = update
    self.delete = delete
    self.batch_insert = batch_insert
    self.batch_update = batch_update


sql_service = SqlService()