stage3_index_report.py

#!/usr/bin/env python3
"""
Generate a Stage 3 index + query-plan report for submission/demo prep.
"""

from __future__ import annotations

import datetime as dt
import os
import sqlite3
from pathlib import Path


ROOT = Path(__file__).resolve().parent
DB_PATH = ROOT / "inventory.db"
OUT_PATH = ROOT / "STAGE3_INDEX_QUERY_MAP.md"


def ensure_stage3_objects(conn: sqlite3.Connection) -> None:
    conn.executescript(
        """
        CREATE TABLE IF NOT EXISTS InventoryTxLog (
            tx_id          INTEGER PRIMARY KEY AUTOINCREMENT,
            tx_type        TEXT    NOT NULL,
            notes          TEXT,
            affected_rows  INTEGER NOT NULL DEFAULT 0,
            created_at     TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );

        CREATE INDEX IF NOT EXISTS idx_products_category_price
            ON Products (category_id, unit_price);
        CREATE INDEX IF NOT EXISTS idx_products_supplier_price
            ON Products (supplier_id, unit_price);
        CREATE INDEX IF NOT EXISTS idx_products_stock_levels
            ON Products (quantity_in_stock, reorder_level);
        CREATE INDEX IF NOT EXISTS idx_products_name
            ON Products (name);
        """
    )
    conn.commit()


def fetch_index_metadata(conn: sqlite3.Connection):
    idx_rows = conn.execute(
        "PRAGMA index_list('Products')"
    ).fetchall()
    out = []
    for row in idx_rows:
        # row tuple shape: seq, name, unique, origin, partial
        idx_name = row[1]
        cols = [r[2] for r in conn.execute(f"PRAGMA index_info('{idx_name}')").fetchall()]
        out.append(
            {
                "name": idx_name,
                "unique": bool(row[2]),
                "origin": row[3],
                "partial": bool(row[4]),
                "columns": cols,
            }
        )
    return out


def explain_plan(conn: sqlite3.Connection, sql: str, params: tuple):
    rows = conn.execute("EXPLAIN QUERY PLAN " + sql, params).fetchall()
    # row tuple shape: id, parent, notused, detail
    return [r[3] for r in rows]


def build_report(conn: sqlite3.Connection) -> str:
    queries = [
        {
            "name": "Dashboard stock alerts",
            "where_used": "app.py -> index() alerts query",
            "sql": """
                SELECT p.product_id, p.name, p.sku
                FROM   Products p
                WHERE  p.quantity_in_stock <= p.reorder_level
                ORDER  BY p.quantity_in_stock ASC
                LIMIT  8
            """,
            "params": (),
        },
        {
            "name": "Product list ordering",
            "where_used": "app.py -> products_list()",
            "sql": """
                SELECT p.product_id, p.name
                FROM   Products p
                ORDER  BY p.name ASC
            """,
            "params": (),
        },
        {
            "name": "Report filter by category + price range",
            "where_used": "app.py -> report()",
            "sql": """
                SELECT p.product_id, p.name
                FROM   Products p
                WHERE  p.category_id = ?
                  AND  p.unit_price >= ?
                  AND  p.unit_price <= ?
                ORDER  BY p.name ASC
            """,
            "params": (1, 10.0, 500.0),
        },
        {
            "name": "Report filter by supplier + price range",
            "where_used": "app.py -> report()",
            "sql": """
                SELECT p.product_id, p.name
                FROM   Products p
                WHERE  p.supplier_id = ?
                  AND  p.unit_price >= ?
                  AND  p.unit_price <= ?
                ORDER  BY p.name ASC
            """,
            "params": (1, 10.0, 500.0),
        },
        {
            "name": "Out-of-stock report slice",
            "where_used": "app.py -> report() stock_status='out_of_stock'",
            "sql": """
                SELECT p.product_id, p.name
                FROM   Products p
                WHERE  p.quantity_in_stock = 0
            """,
            "params": (),
        },
    ]

    indexes = fetch_index_metadata(conn)

    lines = []
    lines.append("# Stage 3 Index and Query Plan Map")
    lines.append("")
    lines.append(f"Generated: {dt.datetime.now().isoformat(timespec='seconds')}")
    lines.append(f"Database: `{DB_PATH}`")
    lines.append("")
    lines.append("## Product Indexes Present")
    lines.append("")
    for idx in indexes:
        lines.append(
            f"- `{idx['name']}` | columns: `{', '.join(idx['columns'])}` | "
            f"unique={idx['unique']} | origin={idx['origin']} | partial={idx['partial']}"
        )
    lines.append("")
    lines.append("## Query Plan Evidence")
    lines.append("")
    for q in queries:
        lines.append(f"### {q['name']}")
        lines.append("")
        lines.append(f"- Used in: `{q['where_used']}`")
        lines.append("- SQL:")
        lines.append("")
        lines.append("```sql")
        lines.append("\n".join(line.rstrip() for line in q["sql"].strip().splitlines()))
        lines.append("```")
        lines.append("")
        lines.append(f"- Params: `{q['params']}`")
        lines.append("- EXPLAIN QUERY PLAN output:")
        lines.append("")
        plans = explain_plan(conn, q["sql"], q["params"])
        lines.append("```text")
        if plans:
            for p in plans:
                lines.append(p)
        else:
            lines.append("(no plan rows returned)")
        lines.append("```")
        lines.append("")

    lines.append("## Notes")
    lines.append("")
    lines.append("- `SEARCH ... USING INDEX ...` indicates indexed lookup.")
    lines.append("- `SCAN` indicates table/index scans; may still be efficient for small tables.")
    lines.append("- These plans can change as table size and data distribution change.")
    lines.append("")

    return "\n".join(lines)


def main() -> None:
    if not DB_PATH.exists():
        raise FileNotFoundError(f"Database not found: {DB_PATH}")

    with sqlite3.connect(DB_PATH) as conn:
        ensure_stage3_objects(conn)
        report = build_report(conn)

    OUT_PATH.write_text(report, encoding="utf-8")
    print(f"Wrote {OUT_PATH}")


if __name__ == "__main__":
    main()