Spaces:
Running on Zero
Running on Zero
File size: 1,925 Bytes
f2a3a7b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | """
SQL executor for running queries against Spider's SQLite databases.
"""
import sqlite3
import threading
import time
def execute_sqlite_query(sql: str, db_path: str, timeout: int = 30) -> dict:
"""
Execute a SQL query against a SQLite database file.
Returns a dict with:
- "success": bool
- "columns": list of column names
- "rows": list of tuples
- "row_count": number of rows
- "execution_time_ms": time taken
- "error": error message (if failure)
"""
conn = None
try:
conn = sqlite3.connect(db_path)
conn.execute("PRAGMA foreign_keys = ON")
conn.text_factory = str
cursor = conn.cursor()
# Enforce hard timeout via conn.interrupt() — the timeout parameter
# in sqlite3.connect() only covers lock acquisition, not query execution.
timer = threading.Timer(timeout, conn.interrupt)
timer.start()
start = time.time()
statements = [s.strip() for s in sql.split(";") if s.strip()]
columns = []
rows = []
try:
for stmt in statements:
cursor.execute(stmt)
if cursor.description:
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
finally:
timer.cancel()
elapsed_ms = round((time.time() - start) * 1000, 2)
return {
"success": True,
"columns": columns,
"rows": rows,
"row_count": len(rows),
"execution_time_ms": elapsed_ms,
"error": None,
}
except Exception as e:
return {
"success": False,
"columns": [],
"rows": [],
"row_count": 0,
"execution_time_ms": 0,
"error": str(e),
}
finally:
if conn:
conn.close()
|