AIが作成したSQLを手計算したSQLite結果と照合する

import sqlite3
from pathlib import Path

SOURCE_DIR = Path("outputs") / "ai_sql_demo"
DATABASE = SOURCE_DIR / "orders.sqlite"

CUSTOMERS = [
    (1, "Ana"),
    (2, "Ben"),
    (3, "Cara"),
]

ORDERS = [
    (101, 1, "2026-08-03", "PAID", 40),
    (102, 1, "2026-08-20", "PAID", 60),
    (103, 1, "2026-08-25", "CANCELLED", 30),
    (104, 2, "2026-08-10", "PAID", 25),
    (105, 2, "2026-09-01", "PAID", 50),
    (106, 3, "2026-08-12", "CANCELLED", 80),
]

SOURCE_DIR.parent.mkdir(parents=True, exist_ok=True)
SOURCE_DIR.mkdir()  # Stop if the synthetic source already exists.

connection = sqlite3.connect(DATABASE)
try:
    connection.executescript("""
        CREATE TABLE customers (
            customer_id INTEGER PRIMARY KEY,
            customer_name TEXT NOT NULL
        );

        CREATE TABLE orders (
            order_id INTEGER PRIMARY KEY,
            customer_id INTEGER NOT NULL,
            order_date TEXT NOT NULL,
            status TEXT NOT NULL,
            amount INTEGER NOT NULL,
            FOREIGN KEY (customer_id) REFERENCES customers(customer_id)
        );
    """)
    connection.executemany(
        "INSERT INTO customers VALUES (?, ?)", CUSTOMERS
    )
    connection.executemany(
        "INSERT INTO orders VALUES (?, ?, ?, ?, ?)", ORDERS
    )
    connection.commit()
finally:
    connection.close()

python create_ai_sql_demo.py

SELECT
    c.customer_id,
    c.customer_name,
    SUM(o.amount) AS august_paid_total
FROM customers AS c
JOIN orders AS o
    ON o.customer_id = c.customer_id
WHERE o.status = 'PAID'
  AND o.order_date >= '2026-08-01'
  AND o.order_date < '2026-09-01'
GROUP BY c.customer_id, c.customer_name
ORDER BY c.customer_id;

SELECT
    c.customer_id,
    c.customer_name,
    COALESCE(SUM(o.amount), 0) AS august_paid_total
FROM customers AS c
LEFT JOIN orders AS o
    ON o.customer_id = c.customer_id
   AND o.status = 'PAID'
   AND o.order_date >= '2026-08-01'
   AND o.order_date < '2026-09-01'
GROUP BY c.customer_id, c.customer_name
ORDER BY c.customer_id;

import csv
import sqlite3
from pathlib import Path

DATABASE = Path("outputs") / "ai_sql_demo" / "orders.sqlite"
OUTPUT_DIR = Path("outputs") / "ai_sql_check_result"
REPORT = OUTPUT_DIR / "sql_check.csv"

AI_SQL = """
SELECT c.customer_id, c.customer_name, SUM(o.amount)
FROM customers AS c
JOIN orders AS o ON o.customer_id = c.customer_id
WHERE o.status = 'PAID'
  AND o.order_date >= '2026-08-01'
  AND o.order_date < '2026-09-01'
GROUP BY c.customer_id, c.customer_name
ORDER BY c.customer_id;
"""

REVISED_SQL = """
SELECT c.customer_id, c.customer_name, COALESCE(SUM(o.amount), 0)
FROM customers AS c
LEFT JOIN orders AS o
    ON o.customer_id = c.customer_id
   AND o.status = 'PAID'
   AND o.order_date >= '2026-08-01'
   AND o.order_date < '2026-09-01'
GROUP BY c.customer_id, c.customer_name
ORDER BY c.customer_id;
"""

EXPECTED = [
    (1, "Ana", 100),
    (2, "Ben", 25),
    (3, "Cara", 0),
]


def run_query(connection, sql):
    return connection.execute(sql).fetchall()


def main() -> None:
    if not DATABASE.is_file():
        raise FileNotFoundError(f"Database not found: {DATABASE}")
    if OUTPUT_DIR.exists():
        raise FileExistsError(f"Output folder already exists: {OUTPUT_DIR}")

    connection = sqlite3.connect(f"file:{DATABASE.resolve()}?mode=ro", uri=True)
    try:
        ai_rows = run_query(connection, AI_SQL)
        revised_rows = run_query(connection, REVISED_SQL)
    finally:
        connection.close()

    OUTPUT_DIR.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT_DIR.mkdir()

    with REPORT.open("x", encoding="utf-8", newline="") as stream:
        writer = csv.writer(stream)
        writer.writerow(["query", "rows", "matches_expected"])
        writer.writerow(["AI", repr(ai_rows), ai_rows == EXPECTED])
        writer.writerow(["REVISED", repr(revised_rows), revised_rows == EXPECTED])

    print(f"Expected rows: {len(EXPECTED)}.")
    print(f"AI rows: {len(ai_rows)}; matches expected: {ai_rows == EXPECTED}.")
    print(
        f"Revised rows: {len(revised_rows)}; "
        f"matches expected: {revised_rows == EXPECTED}."
    )
    print(f"Report: {REPORT.as_posix()}")

    if revised_rows != EXPECTED:
        raise RuntimeError("Revised SQL does not match the expected result.")


if __name__ == "__main__":
    main()

python ai_sql_check.py

Expected rows: 3.
AI rows: 2; matches expected: False.
Revised rows: 3; matches expected: True.
Report: outputs/ai_sql_check_result/sql_check.csv