SQLクエリの生成と検証
SELECT専用モードやパラメータ化クエリなど、安全なSQLのプロンプトパターンを学びます。
「SQLクエリの生成と検証」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン3/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
SQL生成の目標
SQLクエリの生成は作業の半分にすぎません。実際のデータベースに対して実行する前に、クエリが安全で構文的に正しく、ユーザーの意図どおりに動作することを検証する必要があります。
このレッスンでは、SELECTのみの強制、パース、安全な実行、実行計画の検証について説明します。
SELECTのみモードの強制
NL-to-SQLエージェントが実行できる最も危険な処理は、破壊的なステートメントを実行することです。LLMの返す内容にかかわらず、常にSELECTのみモードを強制してください。
単純な文字列チェックでは不十分です。適切なSQLパーサーを使用してください。
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多層防御としてのキーワードブロックリスト
sqlparseを使用していても、二次防御としてキーワードのブロックリストを追加してください。一部のSQLインジェクションはパーサーを欺く可能性があります。実行前に危険なキーワードをチェックすると、安全性をさらに高められます。
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 TruesqlparseによるSQLのパース
sqlparseは、SQL文字列を実行せずにトークン化してパースします。クエリの構造を調べたり、テーブル名を抽出したり、構文上の問題を確認したりできます。
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']テーブルがスキーマに存在することの確認
生成されたSQLからテーブル名を抽出したら、既知のスキーマと照合してください。LLMがテーブル名を誤って生成した場合は、分かりにくいデータベースエラーになる前に、実行せずクエリを拒否します。
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)パラメータ化された実行
ユーザーが提供した値をSQLに挿入するために、文字列フォーマットを決して使用しないでください。クエリを生成するのがLLMであっても、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)実行前のEXPLAINプラン
大規模なテーブルに対する高コストなクエリでは、実際のクエリを実行する前にEXPLAINを実行してください。プランナーが100万行のテーブルに対する全表スキャンを示した場合は、ユーザーに警告するか、クエリを拒否します。
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"]}')行数制限の強制
LLMは、LIMITなしでSELECT * FROM logsを生成することがあり、数百万行を返す可能性があります。クエリにLIMITを追加するか、取得する結果セットに上限を設けることで、常に最大行数を強制してください。
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 500LLM出力からのクリーンなSQLの抽出
LLMはSQLをMarkdownのコードブロック（```sql ... ```）で囲んだり、説明文を付けたりして返すことがよくあります。パースまたは実行する前に、生のSQLを抽出する必要があります。
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))
完全な検証パイプライン
すべての検証手順を1つの関数にまとめ、LLMの生の出力を受け取って、安全に実行できるSQL文字列を返すか、復旧に役立つ説明的なメッセージとともにエラーを発生させるようにします。
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)読み取り専用データベースユーザー
コードレベルの検証は重要ですが、それだけでは不十分です。最後の防御層として、SELECT権限だけを持つ読み取り専用ユーザーアカウントを使ってデータベースに接続してください。悪意のあるクエリがすべてのチェックをすり抜けても、データベースによって拒否されます。
# 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')
)理解度チェック
NL-to-SQLエージェントでSQLを検証する際の、正しい多層防御アプローチは何ですか？
まとめ：SQLの生成と検証
安全なSQL生成には、完全な検証パイプラインが必要です。LLM出力からクリーンなSQLを抽出し、sqlparseを使用してSELECTのみを強制し、キーワードのブロックリストを適用し、テーブル名を実際のスキーマと照合し、行数制限を強制し、最後の安全策として読み取り専用のデータベースユーザーを使用します。
ユーザーが提供した値を扱う場合、パラメータ化クエリによってインジェクションから保護できます。EXPLAINプランのチェックにより、予想外に高コストなクエリが本番データに対して実行されるのを防げます。
よくある質問
「SQLクエリの生成と検証」レッスンは無料ですか？
はい。「SQLクエリの生成と検証」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「SQLクエリの生成と検証」で何を学びますか？
SELECT専用モードやパラメータ化クエリなど、安全なSQLのプロンプトパターンを学びます。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン3/4です。
「SQLクエリの生成と検証」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- NL-to-SQLエージェントの仕組み
- スキーマの理解と注入
- SQLクエリの生成と検証
- 曖昧なデータベース質問への対応