Capstone: text-to-SQL analyst

The big idea

A sales manager asks "what did German customers spend in September?" Today that question waits in a queue for an analyst who knows which of 400 tables holds orders and how countries are coded. You will build an assistant that answers it directly: it finds the relevant tables, writes the SQL, checks the SQL is safe, runs it, and explains the result in a sentence, with the query shown so anyone can check it.

The model is good at the middle step, writing SQL. Everything around it is your job, and most of it is about safety. Generated SQL is untrusted code that you are about to run against real data. A wrong query that runs is worse than one that fails, because it produces a confident wrong number in a meeting.

Think of a new analyst on their first day. You do not give them the admin password. You give them a read-only account limited to the tables they need, a data dictionary, a rule that queries time out after 30 seconds, and you ask them to show their SQL with every number. The capstone builds exactly that setup for a model.

How it works

What you need

• Python 3.10 or newer with SQLite (built in). The runnable core uses a three-table database and a scripted model.
• For the real build: a copy of a real database, or a public sample one, with at least a dozen tables; a model API with structured output; and a database account you can restrict.
• Someone who knows the data to write the expected answers for your tests. Text-to-SQL without known answers cannot be evaluated.

Checkpoint 1: describe and retrieve the schema

The model cannot guess that cust_ctry holds two-letter country codes or that cancelled orders have status = 9. Write a short description for every table and important column, with example values and join keys, and keep it in version control next to the schema. This data dictionary is the most valuable part of the project.

Real warehouses have hundreds of tables, far too many for one prompt, and irrelevant tables confuse the model. So retrieve: embed or keyword-index the table descriptions, find the few that match the question ( Why RAG? ), and put only those in the prompt, plus a handful of example question and SQL pairs for common patterns ( Few-shot prompting and examples ). Tables the user should never query, such as salaries, are not in the index at all.

Done when: for 20 test questions, the tables the correct query needs are among the retrieved ones.

Checkpoint 2: generate SQL as structured output

Ask for a structured response, not free text ( Structured outputs ): the SQL, the tables used, and an assumptions field such as "September means 2026-09-01 to 2026-09-30". Tell the model the SQL dialect, the current date, and to use only the tables shown. When a question is ambiguous ("best customers" by revenue or by order count?), the right output is a clarifying question, not a guess.

Done when: every response parses, names its tables, and states its assumptions.

Checkpoint 3: validate before running

Validation catches mistakes cheaply and early. Check that the query is a single statement, starts with SELECT or WITH, and reads only allowed tables. For real systems use a proper SQL parser for your dialect rather than regular expressions, because SQL has many ways to hide a table name. If validation fails, send the reason back to the model once and let it retry; after that, refuse.

Done when: writes, multiple statements and forbidden tables are all refused with a clear reason.

Checkpoint 4: run read-only with limits

Validation is not a security boundary; the database is. Run generated SQL through an account that can only read the allowed tables or views, on a read-only connection or a replica, with a statement timeout and a row limit. Then a query that slips past your validator still cannot change data or read what it should not, and a runaway join cannot take the database down ( Authentication vs authorization ).

Done when: a write, a forbidden table and an endless query each fail at the database layer, even with the validator turned off.

Checkpoint 5: explain and test

Show the SQL with every answer, then a one-sentence explanation from the model that states the assumptions ("Total of orders placed 1 to 30 September 2026 by customers with country DE: 280 dollars"). Test two kinds of cases: questions with known correct answers, compared on the result rows rather than the SQL text, because many different queries are correct; and unsafe or impossible requests that must fail safely.

Try it

The analyst core: schema retrieval over table descriptions, scripted SQL for five questions, validation, and a read-only connection with a row limit, a step guard and an authorizer that hides the salaries table.

# Text-to-SQL analyst core: retrieve schema, generate SQL (scripted stand-in for a model),
# validate it, run it read-only with limits. Standard library only (sqlite3).
import re, sqlite3

db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, country TEXT);
CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, total REAL, placed_on TEXT);
CREATE TABLE staff_salaries (staff_id INTEGER, salary REAL);
INSERT INTO customers VALUES (1,'Ana','PT'),(2,'Ben','DE'),(3,'Chloe','DE');
INSERT INTO orders VALUES (1,1,120,'2026-09-02'),(2,2,80,'2026-09-15'),(3,3,200,'2026-09-20'),(4,2,40,'2026-10-01');
""")
db.execute("PRAGMA query_only = ON")  # this connection can no longer write
ALLOWED = {"customers": "customers: id, name, country (two-letter code)",
           "orders": "orders: id, customer_id -> customers.id, total in dollars, placed_on date"}
HIDDEN = {r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type = 'table'")} - set(ALLOWED)
db.set_authorizer(lambda action, table, *_:  # the database itself refuses to read hidden tables
                  sqlite3.SQLITE_DENY if action == sqlite3.SQLITE_READ and table in HIDDEN else sqlite3.SQLITE_OK)

def retrieve_schema(question):  # pick tables whose description shares a word with the question
    words = set(re.findall(r"[a-z]+", question.lower()))
    hits = [t for t, d in ALLOWED.items() if words & set(re.findall(r"[a-z]+", d.split(":")[0]))]
    return hits or list(ALLOWED)  # only these descriptions go into the model's prompt

FAKE_SQL = {  # what a model might write for each question, given the schema
    "How many customers are in each country?":
        "SELECT country, COUNT(*) AS customers FROM customers GROUP BY country ORDER BY 2 DESC",
    "What did customers in DE spend on orders in September 2026?":
        "SELECT SUM(o.total) FROM orders o JOIN customers c ON c.id = o.customer_id "
        "WHERE c.country = 'DE' AND o.placed_on BETWEEN '2026-09-01' AND '2026-09-30'",
    "Delete the test orders": "DELETE FROM orders WHERE total < 50",
    "What is the average staff salary?": "SELECT AVG(salary) FROM staff_salaries",
    "List every order": "SELECT * FROM orders; DROP TABLE customers",
}

def validate(sql):  # returns an error message, or None if the query may run
    if ";" in sql.strip().rstrip(";"):
        return "only one statement is allowed"
    if not re.match(r"(?is)^\s*(select|with)\b", sql):
        return "only SELECT queries are allowed"
    ctes = set(re.findall(r"(?i)(?:with(?:\s+recursive)?|,)\s+([a-z_]+)\s*(?:\([^)]*\))?\s+as\s*\(", sql))
    bad = set(re.findall(r"(?i)\b(?:from|join)\s+([a-z_]+)", sql)) - set(ALLOWED) - ctes
    if bad:
        return f"table not allowed: {', '.join(sorted(bad))}"
    return None

def run(sql, max_rows=100, max_steps=100_000):
    steps = [0]
    def guard():  # called by SQLite every 1,000 VM steps; non-zero aborts the query
        steps[0] += 1000
        return steps[0] > max_steps
    db.set_progress_handler(guard, 1000)
    try:
        return db.execute(f"SELECT * FROM ({sql}) LIMIT {max_rows}").fetchall()
    except sqlite3.DatabaseError as e:
        return f"error: {e}"
    finally:
        db.set_progress_handler(None, 0)

def ask(question):
    schema = retrieve_schema(question)
    sql = FAKE_SQL[question]
    problem = validate(sql)
    if problem:
        return f"refused ({problem})"
    return f"{run(sql)}  (schema shown: {', '.join(schema)})"

if __name__ == "__main__":
    for q in FAKE_SQL:
        print(f"Q: {q}\n   {ask(q)}")

Q: How many customers are in each country?
   [('DE', 2), ('PT', 1)]  (schema shown: customers)
Q: What did customers in DE spend on orders in September 2026?
   [(280.0,)]  (schema shown: customers, orders)
Q: Delete the test orders
   refused (only SELECT queries are allowed)
Q: What is the average staff salary?
   refused (table not allowed: staff_salaries)
Q: List every order
   refused (only one statement is allowed)

And the tests, which check answers by result and probe every safety layer, including one the validator misses:

# Tests for the analyst: right answers on known questions, and safe failure on bad SQL.
import os, runpy

app = runpy.run_path(os.path.join(os.path.dirname(os.path.abspath(__file__)), "AI-15.03-analyst.py"))
validate, run, db = app["validate"], app["run"], app["db"]

CORRECT = [  # SQL a model produced -> the answer a person checked by hand
    ("SELECT COUNT(*) FROM orders WHERE placed_on >= '2026-10-01'", [(1,)]),
    ("SELECT name FROM customers WHERE country = 'DE' ORDER BY name", [("Ben",), ("Chloe",)]),
    ("SELECT ROUND(AVG(total), 2) FROM orders", [(110.0,)]),
]
for sql, want in CORRECT:
    got = run(sql)
    print("PASS" if got == want else "FAIL", "answer  ", got)

UNSAFE = ["UPDATE orders SET total = 0", "SELECT * FROM staff_salaries",
          "SELECT 1; DELETE FROM orders", "PRAGMA query_only = OFF"]
for sql in UNSAFE:
    print("PASS" if validate(sql) else "FAIL", "blocked ", validate(sql))

# A runaway query that passes validation is stopped by the step guard.
loop = "WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n) SELECT COUNT(*) FROM n"
print("PASS" if validate(loop) is None and "interrupted" in str(run(loop)) else "FAIL", "runaway ", run(loop))

# A regex validator can be fooled: quoting the table name hides it. The authorizer still says no.
sneaky = 'SELECT AVG(salary) FROM "staff_salaries"'
print("PASS" if "prohibited" in str(run(sneaky)) else "FAIL", "sneaky   validator said", validate(sneaky), "->", run(sneaky))

# Defense in depth: even if a write slipped past the validator, the connection refuses it.
try:
    db.execute("WITH x AS (SELECT 1) DELETE FROM orders")
    print("FAIL", "write ran")
except Exception as e:
    print("PASS", "read-only", e)

PASS answer   [(1,)]
PASS answer   [('Ben',), ('Chloe',)]
PASS answer   [(110.0,)]
PASS blocked  only SELECT queries are allowed
PASS blocked  table not allowed: staff_salaries
PASS blocked  only one statement is allowed
PASS blocked  only SELECT queries are allowed
PASS runaway  error: interrupted
PASS sneaky   validator said None -> error: access to staff_salaries.salary is prohibited
PASS read-only attempt to write a readonly database

1. Look at the "sneaky" test. Why did the regular-expression validator let it through, and which layer stopped it? What does that tell you about where your security boundary should be?
2. "What did customers in DE spend on orders in September 2026?" returns 280. Check it by hand from the inserted rows. Which assumption did the query make about the month, and where should the user see it?
3. Add a question whose correct answer is a clarifying question, such as "Who are our best customers?". What should the test expect?

Going deeper

Measure each stage

A failure can come from retrieval (the right table was not in the prompt), generation (wrong join or filter), or the data (the column means something else than its name suggests). Score them separately: table recall for retrieval, whether the query runs, and whether its rows match the expected rows. A query that runs and returns zero rows is a common silent failure, often a filter value the model invented.

Semantic layers

Many teams stop letting the model write raw SQL over raw tables for important metrics. Instead they define metrics such as revenue or active users once, in a semantic layerA curated layer of business definitions, such as metrics, dimensions and joins, that sits between users and raw tables so everyone computes a number the same way. or in views, and the model picks metrics and filters from that layer. Fewer choices means fewer ways to be wrong, and the number in the chat matches the number on the dashboard.

Cost, latency and caching

Schema retrieval keeps prompts short. Popular questions repeat, so cache by normalized question plus data freshness, and put the long, stable part of the prompt (instructions and examples) first so prompt caching applies ( Latency and cost budgets ). Large scans cost money in cloud warehouses, so use the warehouse's query cost estimate as one more limit.

Final checklist

• A data dictionary with descriptions, example values and join keys, and retrieval over it.
• Structured output with SQL, tables used, assumptions, or a clarifying question.
• Validation with a real parser: single statement, read-only, allowed tables only, one retry with the reason.
• A read-only role limited to allowed tables or views, a statement timeout and a row limit.
• Answers that show the SQL and state assumptions.
• Tests on result rows for known questions, plus unsafe and ambiguous requests that must fail safely.

Trade-offs

Common mistakes

• Running generated SQL with the application's normal database account, which can write.
• Treating the validator as the security boundary instead of database permissions.
• Comparing generated SQL text to a reference query instead of comparing result rows.
• Hiding the SQL from users, so nobody can spot a wrong filter or date range.
• No row or time limits, so one cross join scans the warehouse.

In the wild

Check yourself

Recap