Function execute_query

app/utils/storage.py:133–164

(path: str, sql: str, write_mode: bool = False)

def execute_query(path: str, sql: str, write_mode: bool = False) -> dict:
    conn = open_db(path)
    cursor = conn.cursor()

    if not write_mode:
        cursor.execute("PRAGMA query_only = ON")

    start = time.time()
    cursor.execute(sql)
    elapsed = time.time() - start

    if cursor.description:
        columns = [desc[0] for desc in cursor.description]
        rows = [dict(row) for row in cursor.fetchmany(1000)]
    else:
        columns = []
        rows = []

    affected = cursor.rowcount if not cursor.description else 0

    if write_mode:
        conn.commit()

    conn.close()

    return {
        "columns": columns,
        "rows": rows,
        "total": len(rows),
        "time": elapsed,
        "affected": affected,
    }