storage/local_db.py

from __future__ import annotations
import sqlite3
from typing import Any, Dict, List, Optional

_SCHEMA = """
CREATE TABLE IF NOT EXISTS sessions (
    session_id TEXT PRIMARY KEY, quest_name TEXT NOT NULL, score INTEGER NOT NULL,
    total INTEGER NOT NULL, xp_earned INTEGER NOT NULL, grade TEXT NOT NULL,
    created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS mastery (
    topic TEXT PRIMARY KEY, correct INTEGER NOT NULL DEFAULT 0,
    total INTEGER NOT NULL DEFAULT 0, updated_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS quests (
    quest_id TEXT PRIMARY KEY, name TEXT NOT NULL, topics TEXT NOT NULL,
    boss_topic TEXT NOT NULL, difficulty TEXT NOT NULL,
    unlocked INTEGER NOT NULL DEFAULT 1, completed INTEGER NOT NULL DEFAULT 0,
    created_at TEXT DEFAULT (datetime('now'))
);
"""

class DB:
    def __init__(self, path: str = "studywithchampai.db"):
        self._conn = sqlite3.connect(path, check_same_thread=False)
        self._conn.row_factory = sqlite3.Row
        self._conn.executescript(_SCHEMA)
        self._conn.commit()

    def close(self): self._conn.close()

    def list_tables(self) -> List[str]:
        return [row[0] for row in self._conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'").fetchall()]

    def save_session(self, session_id: str, quest_name: str, score: int,
                     total: int, xp_earned: int, grade: str) -> None:
        self._conn.execute(
            "INSERT OR REPLACE INTO sessions (session_id,quest_name,score,total,xp_earned,grade) VALUES (?,?,?,?,?,?)",
            (session_id, quest_name, score, total, xp_earned, grade))
        self._conn.commit()

    def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
        row = self._conn.execute("SELECT * FROM sessions WHERE session_id=?", (session_id,)).fetchone()
        return dict(row) if row else None

    def upsert_mastery(self, topic: str, correct: int, total: int) -> None:
        self._conn.execute(
            """INSERT INTO mastery (topic,correct,total,updated_at) VALUES (?,?,?,datetime('now'))
               ON CONFLICT(topic) DO UPDATE SET
                 correct=correct+excluded.correct, total=total+excluded.total,
                 updated_at=datetime('now')""",
            (topic, correct, total))
        self._conn.commit()

    def get_mastery_row(self, topic: str) -> Optional[Dict[str, Any]]:
        row = self._conn.execute("SELECT * FROM mastery WHERE topic=?", (topic,)).fetchone()
        return dict(row) if row else None

    def all_mastery_rows(self) -> List[Dict[str, Any]]:
        return [dict(r) for r in self._conn.execute("SELECT * FROM mastery").fetchall()]