100 lines
3.6 KiB
Python
100 lines
3.6 KiB
Python
import sqlite3
|
|
import threading
|
|
from dataclasses import dataclass
|
|
|
|
|
|
class TableState:
|
|
"""Live processing state of a cached table (value of ``TableStats.state``)."""
|
|
|
|
LOADING = "loading" # a full load is in progress
|
|
REFRESHING = "refreshing" # an incremental (delta) refresh is in progress
|
|
READY = "ready" # cached and idle
|
|
STALE = "stale" # TTL expired / schedule slot passed — reloads on next access
|
|
ERROR = "error" # the last load failed
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TableStats:
|
|
rows: int
|
|
columns: list[str]
|
|
# Persisted wall-clock of the last actual data write (full load / delta with rows).
|
|
# Survives restarts. Answers "when did the data last change?".
|
|
last_upsert: str | None
|
|
# In-memory (this process) wall-clock of the last time a refresh cycle ran for the
|
|
# table — bumped even when the cycle wrote nothing. Liveness signal; ``None`` until
|
|
# the first cycle runs after start. Answers "is the refresh loop alive?".
|
|
last_refresh: str | None = None
|
|
state: str = TableState.READY
|
|
tracking: str = "static" # "delta" | "ttl" | "schedule" | "static"
|
|
# Most recent load/refresh failure for this table, if any. ``consecutive_failures``
|
|
# resets to 0 on the next success, so > 0 means the table is currently failing.
|
|
last_error: str | None = None
|
|
last_error_at: str | None = None
|
|
consecutive_failures: int = 0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Stats:
|
|
hits: int
|
|
misses: int
|
|
refetches: int
|
|
tables: dict[str, TableStats]
|
|
errors: int = 0 # total load/refresh failures since start
|
|
db_size_bytes: int = 0 # on-disk cache file size (0 in memory mode)
|
|
|
|
|
|
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, states: dict[str, str] | None = None
|
|
) -> Stats:
|
|
states = states or {}
|
|
with self._lock:
|
|
hits, misses, refetches = self.hits, self.misses, self.refetches
|
|
|
|
tables: dict[str, TableStats] = {}
|
|
cached: set[str] = set()
|
|
for table_name, row_count, last_upsert in conn.execute(
|
|
"SELECT table_name, row_count, last_refresh_at FROM _sqlmem_tables"
|
|
).fetchall():
|
|
cached.add(table_name)
|
|
columns = [
|
|
r[0]
|
|
for r in conn.execute(
|
|
"SELECT column_name FROM _sqlmem_columns WHERE table_name = ? ORDER BY column_name",
|
|
(table_name,),
|
|
).fetchall()
|
|
]
|
|
# last_refresh (run/liveness) is filled in by the engine from the
|
|
# in-memory last-run map; only the persisted write time is read here.
|
|
tables[table_name] = TableStats(
|
|
rows=row_count or 0,
|
|
columns=columns,
|
|
last_upsert=last_upsert,
|
|
state=states.get(table_name, TableState.READY),
|
|
)
|
|
|
|
# Surface tables that are mid-first-load (not yet in _sqlmem_tables) or failed.
|
|
for name, state in states.items():
|
|
if name not in cached and state in (TableState.LOADING, TableState.ERROR):
|
|
tables[name] = TableStats(rows=0, columns=[], last_upsert=None, state=state)
|
|
|
|
return Stats(hits=hits, misses=misses, refetches=refetches, tables=tables)
|