0Pricing
AI Agents · Aula

Gerando e validando consultas SQL

Padrões de prompts para SQL seguro: modo somente SELECT e consultas parametrizadas.

Gerando e validando consultas SQL é uma aula grátis de AI Agents no CoddyKit. Esta é a aula 3 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de AI Agents, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de AI Agents inclui 4 aulas no total.

O objetivo da geração de SQL

Gerar uma consulta SQL é apenas metade do trabalho. Antes de executá-la em um banco de dados real, você precisa validar se a consulta é segura, está sintaticamente correta e faz exatamente o que o usuário pretendia.

Esta lição aborda a aplicação do modo somente SELECT, a análise sintática, a execução segura e a verificação do plano de execução.

Aplicação do modo somente SELECT

A ação mais perigosa que um agente NL-to-SQL pode realizar é executar uma instrução destrutiva. Aplique sempre o modo somente SELECT, independentemente do que o LLM retornar.

Uma verificação ingênua de texto não é suficiente — use um analisador sintático SQL adequado.

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 bloqueio de palavras-chave como defesa em profundidade

Mesmo usando sqlparse, adicione uma lista de bloqueio de palavras-chave como defesa secundária. Algumas injeções SQL podem enganar os analisadores sintáticos. Verificar palavras-chave perigosas antes da execução acrescenta uma camada extra de segurança.

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

Analisando SQL com sqlparse

sqlparse transforma e analisa cadeias SQL sem executá-las. Você pode inspecionar a estrutura da consulta, extrair nomes de tabelas e verificar problemas de sintaxe.

Instale com 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']

Verificando a existência das tabelas no esquema

Depois de extrair os nomes das tabelas do SQL gerado, compare-os com o esquema conhecido. Se o LLM tiver inventado o nome de uma tabela, rejeite a consulta antes da execução, em vez de receber uma mensagem de erro indecifrável do banco de dados.

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)

Execução parametrizada

Nunca use formatação de texto para inserir valores fornecidos pelo usuário no SQL. Embora o LLM gere a consulta, quaisquer valores de filtros fornecidos pelo usuário devem ser passados como parâmetros para evitar injeção 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)

Plano EXPLAIN antes da execução

Para consultas custosas em tabelas grandes, execute EXPLAIN antes da consulta propriamente dita. Se o planejador indicar uma varredura completa em uma tabela com um milhão de linhas, avise o usuário ou rejeite a 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"]}')

Aplicação do limite de linhas

Um LLM pode gerar SELECT * FROM logs sem um LIMIT, o que pode retornar milhões de linhas. Aplique sempre uma quantidade máxima de linhas — anexando LIMIT à consulta ou buscando um conjunto de resultados limitado.

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

Extraindo SQL limpo da saída do LLM

Os LLMs frequentemente retornam SQL envolvido em blocos de código Markdown (```sql ... ```) ou acompanhado de texto explicativo. Você precisa extrair o SQL bruto antes de analisá-lo ou executá-lo.

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

Fluxo completo de validação

Encadeie todas as etapas de validação em uma única função que receba a saída bruta do LLM e retorne uma cadeia SQL segura e executável ou gere um erro com uma mensagem descritiva para permitir a recuperação.

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)

Usuário de banco de dados somente leitura

A validação no nível do código é importante, mas não suficiente. Como camada final de defesa, conecte-se ao banco de dados usando uma conta de usuário somente leitura que tenha apenas privilégios de SELECT. Mesmo que uma consulta maliciosa contorne todas as verificações, o banco de dados a rejeitará.

# 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ção de conhecimento

Qual é a abordagem correta de defesa em profundidade para a validação de SQL em um agente NL-to-SQL?

Recapitulação: geração e validação de SQL

A geração segura de SQL exige um fluxo completo de validação: extraia o SQL limpo da saída do LLM, aplique o modo somente SELECT usando sqlparse, aplique uma lista de bloqueio de palavras-chave, verifique os nomes das tabelas em relação ao esquema real, aplique limites de linhas e use um usuário de banco de dados somente leitura como proteção final.

As consultas parametrizadas protegem contra injeção quando há valores fornecidos pelo usuário. As verificações do plano EXPLAIN impedem a execução de consultas inesperadamente custosas em dados de produção.

Perguntas Frequentes

A aula “Gerando e validando consultas SQL” é grátis?

Sim — o texto completo de “Gerando e validando consultas SQL” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de AI Agents, atualize para CoddyKit PRO. O curso de AI Agents inclui 4 aulas no total.

O que vou aprender em “Gerando e validando consultas SQL”?

Padrões de prompts para SQL seguro: modo somente SELECT e consultas parametrizadas. Você pratica AI Agents com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.

Preciso ter experiência prévia para começar AI Agents?

Nenhuma experiência prévia é necessária. AI Agents no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 3 de 4.

Quanto tempo leva a aula “Gerando e validando consultas SQL”?

A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.

Posso escrever e executar código nesta aula de AI Agents?

Sim. Cada aula de AI Agents inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.

Todas as aulas deste curso

  1. Como funcionam os agentes de NL para SQL
  2. Compreensão e injeção de esquemas
  3. Gerando e validando consultas SQL
  4. Lidando com perguntas ambíguas sobre bancos de dados
← Voltar para AI Agents