feat: add SQL retriever — validated read-only query execution

src/sql_retriever.py

@@ -0,0 +1,61 @@
+"""SQL retriever: execute validated queries against the SQLite knowledge base."""
+
+import os
+import sqlite3
+from pathlib import Path
+
+
+def _validate_sql(sql: str) -> bool:
+    """Validate that a SQL string is a safe SELECT query.
+
+    Rejects non-SELECT statements and queries containing semicolons
+    (which could chain dangerous statements).
+    """
+    stripped = sql.strip()
+    if not stripped:
+        return False
+    if ";" in stripped:
+        return False
+    if not stripped.upper().startswith("SELECT"):
+        return False
+    return True
+
+
+def _get_db_path(cfg: dict) -> str:
+    """Resolve the SQLite database path from config."""
+    sql_db_dir = cfg.get("paths", {}).get("sql_db", "sql_db")
+    if not os.path.isabs(sql_db_dir):
+        project_root = Path(__file__).resolve().parent.parent
+        sql_db_dir = os.path.join(str(project_root), sql_db_dir)
+    return os.path.join(sql_db_dir, "knowledge_base.db")
+
+
+def execute_sql_query(sql_query: str, cfg: dict) -> list[dict]:
+    """Execute a validated SELECT query against the knowledge base SQLite DB.
+
+    Args:
+        sql_query: A SQL SELECT query (generated by the LLM).
+        cfg: App config dict.
+
+    Returns:
+        List of row dicts, or empty list on any error.
+    """
+    if not _validate_sql(sql_query):
+        return []
+
+    db_path = _get_db_path(cfg)
+    if not os.path.exists(db_path):
+        return []
+
+    max_rows = cfg.get("sql", {}).get("max_rows", 200)
+
+    try:
+        conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
+        conn.row_factory = sqlite3.Row
+        cursor = conn.execute(sql_query)
+        rows = cursor.fetchmany(max_rows)
+        result = [dict(row) for row in rows]
+        conn.close()
+        return result
+    except Exception:
+        return []

tests/test_sql_retriever.py

@@ -0,0 +1,105 @@
+# tests/test_sql_retriever.py
+import pytest
+import json
+import sqlite3
+
+
+def test_validate_sql_rejects_non_select():
+    from src.sql_retriever import _validate_sql
+    assert _validate_sql("DROP TABLE users") is False
+    assert _validate_sql("INSERT INTO t VALUES (1)") is False
+    assert _validate_sql("UPDATE t SET x=1") is False
+    assert _validate_sql("DELETE FROM t") is False
+
+
+def test_validate_sql_rejects_semicolons():
+    from src.sql_retriever import _validate_sql
+    assert _validate_sql("SELECT 1; DROP TABLE users") is False
+
+
+def test_validate_sql_accepts_select():
+    from src.sql_retriever import _validate_sql
+    assert _validate_sql("SELECT * FROM t") is True
+    assert _validate_sql("  select Country, Year from t WHERE x = 1  ") is True
+    assert _validate_sql("SELECT COUNT(*) FROM t GROUP BY x") is True
+
+
+def test_validate_sql_rejects_empty():
+    from src.sql_retriever import _validate_sql
+    assert _validate_sql("") is False
+    assert _validate_sql("   ") is False
+
+
+def test_execute_sql_query_returns_rows(tmp_path):
+    from src.sql_retriever import execute_sql_query
+
+    # Create a test DB
+    db_path = tmp_path / "sql_db" / "knowledge_base.db"
+    db_path.parent.mkdir()
+    conn = sqlite3.connect(str(db_path))
+    conn.execute("CREATE TABLE t (Country TEXT, Year INTEGER, Score REAL)")
+    conn.execute("INSERT INTO t VALUES ('China', 2005, 4.0)")
+    conn.execute("INSERT INTO t VALUES ('India', 2005, 3.0)")
+    conn.commit()
+    conn.close()
+
+    cfg = {"paths": {"sql_db": str(tmp_path / "sql_db")}, "sql": {"max_rows": 200}}
+    rows = execute_sql_query("SELECT * FROM t WHERE Country = 'China'", cfg)
+    assert len(rows) == 1
+    assert rows[0]["Country"] == "China"
+    assert rows[0]["Score"] == 4.0
+
+
+def test_execute_sql_query_respects_max_rows(tmp_path):
+    from src.sql_retriever import execute_sql_query
+
+    db_path = tmp_path / "sql_db" / "knowledge_base.db"
+    db_path.parent.mkdir()
+    conn = sqlite3.connect(str(db_path))
+    conn.execute("CREATE TABLE t (x INTEGER)")
+    for i in range(100):
+        conn.execute("INSERT INTO t VALUES (?)", (i,))
+    conn.commit()
+    conn.close()
+
+    cfg = {"paths": {"sql_db": str(tmp_path / "sql_db")}, "sql": {"max_rows": 5}}
+    rows = execute_sql_query("SELECT * FROM t", cfg)
+    assert len(rows) == 5
+
+
+def test_execute_sql_query_rejects_non_select(tmp_path):
+    from src.sql_retriever import execute_sql_query
+
+    db_path = tmp_path / "sql_db" / "knowledge_base.db"
+    db_path.parent.mkdir()
+    conn = sqlite3.connect(str(db_path))
+    conn.execute("CREATE TABLE t (x INTEGER)")
+    conn.commit()
+    conn.close()
+
+    cfg = {"paths": {"sql_db": str(tmp_path / "sql_db")}}
+    rows = execute_sql_query("DROP TABLE t", cfg)
+    assert rows == []
+
+
+def test_execute_sql_query_handles_bad_sql(tmp_path):
+    from src.sql_retriever import execute_sql_query
+
+    db_path = tmp_path / "sql_db" / "knowledge_base.db"
+    db_path.parent.mkdir()
+    conn = sqlite3.connect(str(db_path))
+    conn.execute("CREATE TABLE t (x INTEGER)")
+    conn.commit()
+    conn.close()
+
+    cfg = {"paths": {"sql_db": str(tmp_path / "sql_db")}}
+    rows = execute_sql_query("SELECT * FROM nonexistent_table", cfg)
+    assert rows == []
+
+
+def test_execute_sql_query_no_db(tmp_path):
+    from src.sql_retriever import execute_sql_query
+
+    cfg = {"paths": {"sql_db": str(tmp_path / "sql_db_missing")}}
+    rows = execute_sql_query("SELECT 1", cfg)
+    assert rows == []