Add runtime statistics via engine.stats

src/sqlmem/stats.py

@@ -0,0 +1,61 @@
+import sqlite3
+import threading
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class TableStats:
+    rows: int
+    columns: list[str]
+    last_refresh: str
+
+
+@dataclass(frozen=True)
+class Stats:
+    hits: int
+    misses: int
+    refetches: int
+    tables: dict[str, TableStats]
+
+
+class StatsCollector:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self.hits = 0
+        self.misses = 0
+        self.refetches = 0
+
+    def record_hit(self) -> None:
+        with self._lock:
+            self.hits += 1
+
+    def record_miss(self) -> None:
+        with self._lock:
+            self.misses += 1
+
+    def record_refetch(self) -> None:
+        with self._lock:
+            self.refetches += 1
+
+    def snapshot(self, conn: sqlite3.Connection) -> Stats:
+        with self._lock:
+            hits, misses, refetches = self.hits, self.misses, self.refetches
+
+        tables: dict[str, TableStats] = {}
+        for table_name, row_count, last_refresh in conn.execute(
+            "SELECT table_name, row_count, last_refresh_at FROM _sqlmem_tables"
+        ).fetchall():
+            columns = [
+                r[0]
+                for r in conn.execute(
+                    "SELECT column_name FROM _sqlmem_columns WHERE table_name = ? ORDER BY column_name",
+                    (table_name,),
+                ).fetchall()
+            ]
+            tables[table_name] = TableStats(
+                rows=row_count or 0,
+                columns=columns,
+                last_refresh=last_refresh,
+            )
+
+        return Stats(hits=hits, misses=misses, refetches=refetches, tables=tables)