NL2SQL_ENGINE / execution_checker.py
nisar9034's picture
Upload 5 files
5e468f2 verified
import sqlite3
def get_best_query(schema_create_statements, candidate_queries):
"""
Creates an in-memory database, builds the user's tables,
and tests the AI's generated queries to find the first valid one.
"""
# 1. Create a temporary database in RAM (disappears when the function ends)
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()
# 2. Build the empty tables using the user's schema
try:
cursor.executescript(schema_create_statements)
except sqlite3.OperationalError as e:
return f"Error: The provided schema is invalid. ({e})"
# 3. Test the AI's candidate queries
for i, query in enumerate(candidate_queries):
try:
# We use EXPLAIN to check syntax without actually querying data
cursor.execute(f"EXPLAIN {query}")
conn.close()
return f"-- Selected Candidate #{i+1} (Syntax Valid)\n{query}"
except sqlite3.OperationalError as e:
# If there is a syntax error, we ignore it and try the next candidate
print(f"Candidate {i+1} failed syntax check: {e}")
continue
conn.close()
return "Error: All generated queries contained syntax errors."
# --- TESTING BLOCK ---
if __name__ == "__main__":
# The frontend will provide the raw CREATE statements
test_schema = """
CREATE TABLE employees (id INTEGER PRIMARY KEY, name TEXT, salary REAL);
CREATE TABLE departments (id INTEGER PRIMARY KEY, name TEXT);
"""
# The AI generates these. Notice the first two have deliberate syntax errors.
ai_candidates = [
"SELECT SUMM(salary) FROM employees", # Fails: Bad function name
"SELECT sum(salary) FROM employees JOIN bad_table", # Fails: Table doesn't exist
"SELECT sum(salary) FROM employees" # Passes: Perfect SQLite syntax
]
print("Testing AI Candidates against In-Memory DB...\n")
final_output = get_best_query(test_schema, ai_candidates)
print("\nWinning Query to show the user:\n" + final_output)