0Pricing
AI Agents · Lezione

Generazione e convalida delle query SQL

Pattern di prompt per SQL sicuro: modalità con sole SELECT e query parametrizzate

Generazione e convalida delle query SQL è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 3 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.

L'obiettivo della generazione SQL

Generare una query SQL è solo metà del lavoro. Prima di eseguirla su un database reale, deve convalidare che sia sicura, sintatticamente corretta e che faccia esattamente ciò che l'utente intendeva.

Questa lezione tratta l'imposizione delle sole istruzioni SELECT, l'analisi, l'esecuzione sicura e la verifica del piano di esecuzione.

Imposizione della modalità solo SELECT

La cosa più pericolosa che un agente NL-to-SQL può fare è eseguire un'istruzione distruttiva. Imponga sempre la modalità solo SELECT, indipendentemente da ciò che restituisce l'LLM.

Un semplice controllo della stringa non è sufficiente: utilizzi un parser SQL appropriato.

import sqlparse

def is_select_only(sql):
    parsed = sqlparse.parse(sql)
    if not parsed:
        return False
    for statement in parsed:
        stmt_type = statement.get_type()
        if stmt_type != 'SELECT':
            print(f'Blocked statement type: {stmt_type}')
            return False
    return True

# Test
print(is_select_only('SELECT * FROM users'))  # True
print(is_select_only('DROP TABLE users'))      # False — Blocked

Blocklist di parole chiave come difesa in profondità

Anche con sqlparse, aggiunga una blocklist di parole chiave come difesa secondaria. Alcune SQL injection possono ingannare i parser. Controllare la presenza di parole chiave pericolose prima dell'esecuzione aggiunge un ulteriore livello di sicurezza.

DANGEROUS_KEYWORDS = [
    'INSERT', 'UPDATE', 'DELETE', 'DROP', 'CREATE',
    'ALTER', 'TRUNCATE', 'GRANT', 'REVOKE', 'EXEC',
    'EXECUTE', 'CALL', 'MERGE'
]

def passes_blocklist(sql):
    sql_upper = sql.upper()
    for keyword in DANGEROUS_KEYWORDS:
        # Check as whole word to avoid false positives like 'CREATED_AT'
        import re
        if re.search(r'\b' + keyword + r'\b', sql_upper):
            raise ValueError(f'Blocked keyword detected: {keyword}')
    return True

def validate_sql(sql):
    if not is_select_only(sql):
        raise ValueError('Only SELECT statements are allowed')
    passes_blocklist(sql)
    return True

Analisi SQL con sqlparse

sqlparse tokenizza e analizza le stringhe SQL senza eseguirle. Può ispezionare la struttura della query, estrarre i nomi delle tabelle e verificare la presenza di problemi di sintassi.

Installi con pip install sqlparse.

import sqlparse
from sqlparse.sql import IdentifierList, Identifier
from sqlparse.tokens import Keyword, DML

def extract_table_names(sql):
    parsed = sqlparse.parse(sql)[0]
    tables = []
    from_seen = False
    for token in parsed.tokens:
        if token.ttype is DML and token.value.upper() == 'SELECT':
            continue
        if token.ttype is Keyword and token.value.upper() in ('FROM', 'JOIN'):
            from_seen = True
            continue
        if from_seen:
            if isinstance(token, Identifier):
                tables.append(token.get_name())
            elif isinstance(token, IdentifierList):
                for item in token.get_identifiers():
                    tables.append(item.get_name())
            from_seen = False
    return tables

print(extract_table_names('SELECT u.name FROM users u JOIN orders o ON u.id = o.user_id'))
# ['users', 'orders']

Verifica dell'esistenza delle tabelle nello schema

Dopo aver estratto i nomi delle tabelle dal codice SQL generato, li confronti con lo schema noto. Se l'LLM ha allucinato il nome di una tabella, rifiuti la query prima dell'esecuzione invece di ottenere un criptico errore del database.

def validate_tables_exist(sql, known_tables):
    used_tables = extract_table_names(sql)
    invalid = [t for t in used_tables if t and t not in known_tables]
    if invalid:
        raise ValueError(
            f'Query references non-existent tables: {invalid}. '
            f'Available tables: {list(known_tables)[:10]}...'
        )
    return True

# Usage
known = set(build_schema_dict(conn).keys())
try:
    validate_tables_exist(generated_sql, known)
except ValueError as e:
    # Send error back to LLM for correction
    corrected_sql = llm_fix_sql(generated_sql, str(e))
    print('Corrected SQL:', corrected_sql)

Esecuzione parametrizzata

Non utilizzi mai la formattazione delle stringhe per inserire in SQL valori forniti dall'utente. Anche se è l'LLM a generare la query, tutti i valori dei filtri forniti dall'utente devono essere passati come parametri per prevenire le SQL injection.

import sqlite3

conn = sqlite3.connect(':memory:')
conn.execute('CREATE TABLE orders (status TEXT, user_id INTEGER)')
conn.execute("INSERT INTO orders VALUES ('pending', 42)")

def safe_execute(conn, sql_template, params=()):
    """Execute with parameterized values."""
    cur = conn.cursor()
    cur.execute(sql_template, params)  # driver handles escaping
    columns = [d[0] for d in cur.description]
    rows = cur.fetchmany(200)
    return {'columns': columns, 'rows': rows}

sql = 'SELECT * FROM orders WHERE status = ? AND user_id = ?'
result = safe_execute(conn, sql, params=('pending', 42))
print(result)

Piano EXPLAIN prima dell'esecuzione

Per le query costose eseguite su tabelle di grandi dimensioni, esegua EXPLAIN prima della query effettiva. Se il planner mostra una scansione completa di una tabella con un milione di righe, avverta l'utente o rifiuti la query.

def check_explain_plan(conn, sql):
    explain_sql = f'EXPLAIN {sql}'
    with conn.cursor() as cur:
        cur.execute(explain_sql)
        plan = '\n'.join(row[0] for row in cur.fetchall())

    # Check for sequential scans on large tables
    if 'Seq Scan' in plan:
        print('WARNING: Query involves a sequential scan')
        print(plan)
        return {'safe': False, 'plan': plan, 'warning': 'Sequential scan detected'}

    return {'safe': True, 'plan': plan}

# Use before executing
plan_result = check_explain_plan(conn, generated_sql)
if not plan_result['safe']:
    print(f'Optimization hint: {plan_result["warning"]}')

Imposizione del limite di righe

Un LLM potrebbe generare SELECT * FROM logs senza un LIMIT, restituendo potenzialmente milioni di righe. Imponga sempre un numero massimo di righe, aggiungendo LIMIT alla query oppure recuperando un insieme di risultati con un limite.

import re

MAX_ROWS = 500

def enforce_row_limit(sql, max_rows=MAX_ROWS):
    sql_upper = sql.upper().rstrip().rstrip(';')

    # Check if LIMIT already present
    if re.search(r'\bLIMIT\b', sql_upper):
        # Extract current limit and enforce maximum
        match = re.search(r'LIMIT\s+(\d+)', sql_upper)
        if match:
            current = int(match.group(1))
            if current > max_rows:
                sql = re.sub(r'LIMIT\s+\d+', f'LIMIT {max_rows}', sql, flags=re.IGNORECASE)
    else:
        sql = sql.rstrip(';') + f' LIMIT {max_rows}'

    return sql

print(enforce_row_limit('SELECT * FROM users'))
# SELECT * FROM users LIMIT 500

Estrazione del codice SQL pulito dall'output dell'LLM

Gli LLM restituiscono spesso SQL racchiuso in blocchi di codice Markdown (```sql ... ```) o accompagnato da testo esplicativo. Deve estrarre il codice SQL puro prima di analizzarlo o eseguirlo.

import re

CODE_FENCE = chr(96) * 3  # three backticks, built at runtime to avoid template issues

def extract_sql(llm_response):
    # Remove markdown code blocks ('''sql ... ''' or ''' ... ''')
    pattern = CODE_FENCE + r'(?:sql)?\s*([\s\S]+?)' + CODE_FENCE
    match = re.search(pattern, llm_response, re.IGNORECASE)
    if match:
        return match.group(1).strip()

    # If no code block, look for SELECT statement
    match = re.search(r'(SELECT\s+[\s\S]+?;)', llm_response, re.IGNORECASE)
    if match:
        return match.group(1).strip()

    # Fallback: strip common preamble phrases
    cleaned = re.sub(r'^(Here is|The SQL query is|Query:)[^\n]*\n', '',
                     llm_response, flags=re.IGNORECASE).strip()
    return cleaned

if __name__ == '__main__':
    demo_response = 'Here is the SQL query:\n' + CODE_FENCE + 'sql\nSELECT * FROM users;\n' + CODE_FENCE
    print(extract_sql(demo_response))

Pipeline completa di convalida

Riunisca tutti i passaggi di convalida in un'unica funzione che riceva l'output grezzo dell'LLM e restituisca una stringa SQL sicura ed eseguibile, oppure sollevi un errore con un messaggio descrittivo per consentirne la gestione.

def validate_and_prepare_sql(llm_output, known_tables, max_rows=500):
    # Step 1: extract raw SQL
    sql = extract_sql(llm_output)
    if not sql:
        raise ValueError('No SQL found in LLM response')

    # Step 2: type check
    if not is_select_only(sql):
        raise ValueError('Only SELECT queries allowed')

    # Step 3: keyword blocklist
    passes_blocklist(sql)

    # Step 4: table existence check
    validate_tables_exist(sql, known_tables)

    # Step 5: row limit
    sql = enforce_row_limit(sql, max_rows)

    return sql

# Full flow
try:
    safe_sql = validate_and_prepare_sql(llm_output, known_tables)
    result = safe_execute(conn, safe_sql)
except ValueError as e:
    corrected = llm_fix_sql(llm_output, str(e))
    safe_sql = validate_and_prepare_sql(corrected, known_tables)
    result = safe_execute(conn, safe_sql)

Utente del database in sola lettura

La convalida a livello di codice è importante, ma non sufficiente. Come ultimo livello di difesa, si connetta al database utilizzando un account utente in sola lettura che disponga solo dei privilegi SELECT. Anche se una query malevola riesce a eludere tutti i controlli, il database la rifiuterà.

# Create read-only user in PostgreSQL:
# CREATE USER nl_to_sql_reader WITH PASSWORD 'secure_password';
# GRANT CONNECT ON DATABASE yourdb TO nl_to_sql_reader;
# GRANT USAGE ON SCHEMA public TO nl_to_sql_reader;
# GRANT SELECT ON ALL TABLES IN SCHEMA public TO nl_to_sql_reader;

import os
import psycopg2

def get_readonly_connection():
    return psycopg2.connect(
        host=os.getenv('DB_HOST'),
        database=os.getenv('DB_NAME'),
        user='nl_to_sql_reader',       # read-only account
        password=os.getenv('DB_READER_PASS')
    )

Verifica delle conoscenze

Qual è l'approccio corretto di difesa in profondità per la convalida SQL in un agente NL-to-SQL?

Riepilogo: generazione e convalida SQL

La generazione sicura di SQL richiede una pipeline completa di convalida: estrarre il codice SQL pulito dall'output dell'LLM, imporre l'uso delle sole istruzioni SELECT con sqlparse, applicare una blocklist di parole chiave, verificare i nomi delle tabelle rispetto allo schema reale, imporre limiti al numero di righe e utilizzare un utente del database in sola lettura come protezione finale.

Le query parametrizzate proteggono dalle SQL injection quando sono coinvolti valori forniti dall'utente. I controlli del piano EXPLAIN impediscono l'esecuzione di query inaspettatamente costose sui dati di produzione.

Domande Frequenti

La lezione «Generazione e convalida delle query SQL» è gratuita?

Sì — il testo completo di «Generazione e convalida delle query SQL» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.

Cosa imparerò in «Generazione e convalida delle query SQL»?

Pattern di prompt per SQL sicuro: modalità con sole SELECT e query parametrizzate Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare AI Agents?

Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 3 di 4.

Quanto tempo richiede la lezione «Generazione e convalida delle query SQL»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione AI Agents?

Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

  1. Come funzionano gli agenti NL-to-SQL
  2. Comprensione e iniezione dello schema
  3. Generazione e convalida delle query SQL
  4. Gestione delle domande ambigue sui database
← Torna a AI Agents