khalilbenaz/claude-skills-collection/agent-database-query-subagent

required = ["question", "connection.db_type", "connection.host", "connection.database"]
# Tester la connexion : ping + SELECT 1
# Si échec → retourner immédiatement errors=[{"type": "connection_error", ...}]

-- PostgreSQL / MySQL
SELECT table_name, column_name, data_type, is_nullable
FROM information_schema.columns
WHERE table_schema = 'public'
ORDER BY table_name, ordinal_position;
 
-- SQLite
SELECT name, sql FROM sqlite_master WHERE type='table';
 
-- SQL Server
SELECT t.name, c.name, tp.name, c.is_nullable
FROM sys.tables t
JOIN sys.columns c ON t.object_id = c.object_id
JOIN sys.types tp ON c.user_type_id = tp.user_type_id;

Schéma DDL :
<DDL des tables pertinentes uniquement>
 
Question : <question utilisateur>
Dialecte : <db_type>
Contraintes : lecture seule, LIMIT max_rows
 
Règles :
- Préférer les CTEs aux sous-requêtes imbriquées
- Alias explicites sur toutes les colonnes ambiguës
- Pas de SELECT * sur tables volumineuses
- Exemples few-shot si disponibles en session_context

import sqlglot
def validate_query(sql: str, db_type: str, schema: dict, read_only: bool) -> list[str]:
    errors = []
    # 1. Parse syntaxique
    try:
        parsed = sqlglot.parse_one(sql, dialect=db_type)
    except sqlglot.errors.ParseError as e:
        errors.append(f"syntax_error: {e}")
        return errors
    # 2. Vérifier colonnes et tables vs schéma
    for table in parsed.find_all(sqlglot.exp.Table):
        if table.name not in schema["tables"]:
            errors.append(f"unknown_table: {table.name}")
    # 3. Bloquer mutations si read_only
    if read_only:
        forbidden = (sqlglot.exp.Drop, sqlglot.exp.Delete,
                     sqlglot.exp.Update, sqlglot.exp.Insert,
                     sqlglot.exp.Create, sqlglot.exp.AlterTable)
        for node in parsed.walk():
            if isinstance(node, forbidden):
                errors.append(f"mutation_blocked: {type(node).__name__}")
    return errors

from sqlalchemy import create_engine, text
from sqlalchemy.pool import NullPool
engine = create_engine(conn_url, poolclass=NullPool,
                       connect_args={"connect_timeout": 5})
with engine.connect() as conn:
    # Lecture seule au niveau transaction
    conn.execute(text("SET TRANSACTION READ ONLY"))          # PostgreSQL / MySQL
    # SQL Server : utiliser un login sans droits DML
    # Timeout
    conn.execute(text(f"SET statement_timeout = {timeout_s * 1000}"))  # PostgreSQL ms
    # LIMIT automatique si absent
    if "LIMIT" not in sql.upper() and db_type != "sqlserver":
        sql += f" LIMIT {max_rows}"
    elif db_type == "sqlserver" and "TOP" not in sql.upper():
        sql = sql.replace("SELECT", f"SELECT TOP {max_rows}", 1)
    result = conn.execute(text(sql))
    rows = [dict(r) for r in result.fetchmany(max_rows + 1)]
    truncated = len(rows) > max_rows
    return rows[:max_rows], truncated

import math
from decimal import Decimal
from datetime import date, datetime
def serialize_row(row: dict) -> dict:
    out = {}
    for k, v in row.items():
        if v is None:
            out[k] = None
        elif isinstance(v, (datetime, date)):
            out[k] = v.isoformat()
        elif isinstance(v, Decimal):
            out[k] = float(v)
        elif isinstance(v, float) and math.isnan(v):
            out[k] = None  # NaN non serializable JSON
        elif isinstance(v, bytes):
            out[k] = v.hex()
        else:
            out[k] = v
    return out

EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) <votre requête>;  -- PostgreSQL
EXPLAIN FORMAT=JSON <votre requête>;                       -- MySQL

class QuerySession:
    def __init__(self):
        self.history: list[dict] = []  # [{"question", "sql", "row_count", "columns"}]
        self.last_result_columns: list[str] = []
        self.last_filter: dict = {}
    def resolve_anaphora(self, question: str) -> str:
        """Remplacer 'eux', 'les mêmes', 'parmi ceux-là' par le contexte précédent."""
        if self.history and any(p in question.lower() for p in ["eux", "ceux-là", "les mêmes"]):
            last = self.history[-1]
            return f"Parmi les résultats de '{last['question']}' ({last['sql']}), {question}"
        return question

{
  "question": str,           # Obligatoire — NL ou SQL direct
  "schema": dict | None,     # Optionnel — inféré si absent
  "connection": {
    "db_type": str,          # "postgresql"|"mysql"|"sqlite"|"sqlserver"|"bigquery"|"mongodb"
    "host": str,
    "port": int | None,
    "database": str,
    "username": str,
    "password": str,         # Jamais loggué
    "ssl": bool              # Défaut: True
  },
  "read_only": bool,         # Défaut: True
  "max_rows": int,           # Défaut: 1000
  "timeout_s": int,          # Défaut: 30
  "session_context": dict | None,
  "explain_results": bool    # Défaut: True
}

{
  "query": str,
  "results": list[dict],
  "row_count": int,
  "execution_time_s": float,
  "explanation": str,
  "stats": {"col": {"min": float, "max": float, "mean": float, "null_count": int}},
  "optimization_hints": list[str],
  "visualization_suggestion": str,
  "follow_up_questions": list[str],
  "session_context": dict,
  "errors": list[{"type": str, "message": str, "query": str}],
  "truncated": bool
}

sqlalchemy>=2.0
sqlglot>=25.0
psycopg2-binary>=2.9
pymysql>=1.1
pyodbc>=5.0
pandas>=2.2
pydantic>=2.0