0Pricing
AI Agents · Leçon

Générer et valider des requêtes SQL

Schémas de prompts pour un SQL sûr : mode SELECT uniquement et requêtes paramétrées.

Générer et valider des requêtes SQL est une leçon AI Agents gratuite sur CoddyKit. Ceci est la leçon 3 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Agents, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Agents comprend 4 leçons au total.

Objectif de génération de SQL

Générer une requête SQL ne représente que la moitié du travail. Avant de l'exécuter sur une véritable base de données, vous devez valider qu'elle est sûre, correcte du point de vue syntaxique et qu'elle fait exactement ce que l'utilisateur voulait.

Cette leçon traite de l'application du mode SELECT uniquement, de l'analyse syntaxique, de l'exécution sécurisée et de la vérification du plan d'exécution.

Application du mode SELECT uniquement

La chose la plus dangereuse qu'un agent NL vers SQL puisse faire est d'exécuter une instruction destructive. Appliquez toujours le mode SELECT uniquement, quel que soit le résultat renvoyé par le LLM.

Une simple vérification de chaîne ne suffit pas : utilisez un analyseur SQL approprié.

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

Liste de blocage de mots-clés en défense en profondeur

Même avec sqlparse, ajoutez une liste de blocage de mots-clés comme défense secondaire. Certaines injections SQL peuvent tromper les analyseurs. Vérifier la présence de mots-clés dangereux avant l'exécution ajoute une couche de sécurité supplémentaire.

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

Analyse de SQL avec sqlparse

sqlparse segmente et analyse les chaînes SQL sans les exécuter. Vous pouvez examiner la structure de la requête, extraire les noms de tables et vérifier les problèmes de syntaxe.

Installez-le avec 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']

Vérification de l'existence des tables dans le schéma

Après avoir extrait les noms de tables du SQL généré, comparez-les à votre schéma de référence. Si le LLM a halluciné le nom d'une table, rejetez la requête avant son exécution plutôt que d'obtenir une erreur de base de données obscure.

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)

Exécution paramétrée

N'utilisez jamais la mise en forme de chaînes pour injecter dans SQL des valeurs fournies par l'utilisateur. Même si le LLM génère la requête, toutes les valeurs de filtre fournies par l'utilisateur doivent être transmises comme paramètres afin d'empêcher les injections 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 avant l'exécution

Pour les requêtes coûteuses visant de grandes tables, exécutez EXPLAIN avant la requête proprement dite. Si le planificateur indique un parcours complet d'une table d'un million de lignes, avertissez l'utilisateur ou rejetez la requête.

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

Application d'une limite de lignes

Un LLM peut générer SELECT * FROM logs sans LIMIT, ce qui peut renvoyer des millions de lignes. Appliquez toujours un nombre maximal de lignes, soit en ajoutant LIMIT à la requête, soit en récupérant un ensemble de résultats borné.

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

Extraction du SQL propre depuis la sortie du LLM

Les LLM renvoient souvent du SQL entouré de blocs de code Markdown (```sql ... ```) ou accompagné d'un texte explicatif. Vous devez extraire le SQL brut avant de l'analyser ou de l'exécuter.

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

Chaîne complète de validation

Enchaînez toutes les étapes de validation dans une seule fonction qui accepte la sortie brute du LLM et renvoie une chaîne SQL sûre et exécutable ou lève une erreur accompagnée d'un message explicite pour permettre la reprise.

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)

Utilisateur de base de données en lecture seule

La validation au niveau du code est importante, mais elle ne suffit pas. Comme dernière ligne de défense, connectez-vous à la base de données avec un compte utilisateur en lecture seule qui ne possède que les privilèges SELECT. Même si une requête malveillante contourne tous les contrôles, la base de données la rejettera.

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

Vérification des connaissances

Quelle est la bonne approche de défense en profondeur pour valider le SQL dans un agent NL vers SQL ?

Récapitulatif : génération et validation de SQL

La génération sûre de SQL exige une chaîne complète de validation : extraire un SQL propre de la sortie du LLM, appliquer le mode SELECT uniquement à l'aide de sqlparse, appliquer une liste de blocage de mots-clés, vérifier les noms de tables par rapport au schéma réel, appliquer des limites de lignes et utiliser un utilisateur de base de données en lecture seule comme dernière protection.

Les requêtes paramétrées protègent contre les injections lorsque des valeurs fournies par l'utilisateur sont utilisées. Les vérifications du plan EXPLAIN empêchent l'exécution de requêtes anormalement coûteuses sur les données de production.

Questions Fréquemment Posées

La leçon « Générer et valider des requêtes SQL » est-elle gratuite ?

Oui — le texte complet de « Générer et valider des requêtes SQL » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Agents, passe à CoddyKit PRO. Le cours AI Agents comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Générer et valider des requêtes SQL » ?

Schémas de prompts pour un SQL sûr : mode SELECT uniquement et requêtes paramétrées. Tu pratiques AI Agents avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer AI Agents ?

Aucune expérience préalable n'est requise. AI Agents sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 3 sur 4.

Combien de temps prend la leçon « Générer et valider des requêtes SQL » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon AI Agents ?

Oui. Chaque leçon AI Agents inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Fonctionnement des agents NL-to-SQL
  2. Comprendre et injecter un schéma
  3. Générer et valider des requêtes SQL
  4. Gérer les questions ambiguës sur les bases de données
← Retour à AI Agents