0Pricing
AI Agents · Lección

Generación y validación de consultas SQL

Patrones de prompts para SQL seguro: modo de solo SELECT y consultas parametrizadas.

Generación y validación de consultas SQL es una lección gratuita de AI Agents en CoddyKit. Esta es la lección 3 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Agents, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Agents incluye 4 lecciones en total.

El objetivo de generar SQL

Generar una consulta SQL es solo la mitad del trabajo. Antes de ejecutarla en una base de datos real, debe validar que la consulta sea segura, sintácticamente correcta y que haga exactamente lo que el usuario pretendía.

Esta lección aborda la aplicación del modo exclusivo de SELECT, el análisis sintáctico, la ejecución segura y la verificación del plan de ejecución.

Aplicación del modo exclusivo de SELECT

Lo más peligroso que puede hacer un agente NL-to-SQL es ejecutar una instrucción destructiva. Aplique siempre el modo exclusivo de SELECT, independientemente de lo que devuelva el LLM.

Una comprobación sencilla de cadenas no es suficiente: utilice un analizador sintáctico de SQL adecuado.

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

Lista de bloqueo de palabras clave como defensa en profundidad

Incluso con sqlparse, añada una lista de bloqueo de palabras clave como defensa secundaria. Algunas inyecciones SQL pueden engañar a los analizadores sintácticos. Comprobar la presencia de palabras clave peligrosas antes de la ejecución añade una capa adicional de seguridad.

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

Análisis sintáctico de SQL con sqlparse

sqlparse tokeniza y analiza cadenas SQL sin ejecutarlas. Puede inspeccionar la estructura de la consulta, extraer nombres de tablas y comprobar si hay problemas de sintaxis.

Instálelo 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']

Verificación de la existencia de tablas en el esquema

Después de extraer los nombres de las tablas del SQL generado, compárelos con el esquema conocido. Si el LLM ha inventado el nombre de una tabla, rechace la consulta antes de ejecutarla en lugar de obtener un error críptico de la base de datos.

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)

Ejecución parametrizada

No utilice nunca el formateo de cadenas para insertar en SQL valores proporcionados por el usuario. Aunque el LLM genere la consulta, todos los valores de filtros proporcionados por el usuario deben pasarse como parámetros para evitar la inyección SQL.

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)

Plan EXPLAIN antes de la ejecución

Para consultas costosas en tablas grandes, ejecute EXPLAIN antes de la consulta real. Si el planificador muestra un escaneo completo de una tabla con un millón de filas, advierta al usuario o rechace la consulta.

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"]}')

Aplicación de un límite de filas

Un LLM podría generar SELECT * FROM logs sin un LIMIT, lo que podría devolver millones de filas. Aplique siempre un número máximo de filas: añada LIMIT a la consulta o recupere un conjunto de resultados acotado.

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

Extracción de SQL limpio de la salida del LLM

A menudo, los LLM devuelven SQL dentro de bloques de código Markdown (```sql ... ```) o acompañado de texto explicativo. Debe extraer el SQL sin formato antes de analizarlo sintácticamente o ejecutarlo.

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))

Flujo completo de validación

Encadene todos los pasos de validación en una sola función que reciba la salida sin formato del LLM y devuelva una cadena SQL segura y ejecutable, o genere un error con un mensaje descriptivo para facilitar la recuperación.

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)

Usuario de base de datos de solo lectura

La validación a nivel de código es importante, pero no suficiente. Como última capa de defensa, conéctese a la base de datos mediante una cuenta de usuario de solo lectura que tenga únicamente privilegios de SELECT. Aunque una consulta maliciosa eluda todas las comprobaciones, la base de datos la rechazará.

# 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')
    )

Comprobación de conocimientos

¿Cuál es el enfoque correcto de defensa en profundidad para validar SQL en un agente NL-to-SQL?

Resumen: generación y validación de SQL

La generación segura de SQL requiere un flujo completo de validación: extraer SQL limpio de la salida del LLM, aplicar el modo exclusivo de SELECT mediante sqlparse, utilizar una lista de bloqueo de palabras clave, verificar los nombres de las tablas con el esquema real, aplicar límites de filas y usar un usuario de base de datos de solo lectura como protección final.

Las consultas parametrizadas protegen contra la inyección cuando intervienen valores proporcionados por el usuario. Las comprobaciones del plan EXPLAIN evitan que se ejecuten consultas inesperadamente costosas sobre datos de producción.

Preguntas frecuentes

¿La lección «Generación y validación de consultas SQL» es gratis?

Sí — el texto completo de «Generación y validación de consultas SQL» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Agents, actualiza a CoddyKit PRO. El curso de AI Agents incluye 4 lecciones en total.

¿Qué aprenderé en «Generación y validación de consultas SQL»?

Patrones de prompts para SQL seguro: modo de solo SELECT y consultas parametrizadas. Practicas AI Agents con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.

¿Necesito experiencia previa para empezar AI Agents?

No se requiere experiencia previa. AI Agents en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 3 de 4.

¿Cuánto tiempo toma la lección «Generación y validación de consultas SQL»?

La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.

¿Puedo escribir y ejecutar código en esta lección de AI Agents?

Sí. Cada lección de AI Agents incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.

Todas las lecciones de este curso

  1. Cómo funcionan los agentes NL-to-SQL
  2. Comprensión e inyección de esquemas
  3. Generación y validación de consultas SQL
  4. Gestión de preguntas ambiguas sobre bases de datos
← Volver a AI Agents