SQLite Dashboard Caching

You have a dashboard. It hits SQLite with the same expensive queries every time someone refreshes the page. GROUP BY on thousands of rows. JOINs across five tables. Window functions that take 200ms a pop.

With inhouse, you cache those query results in memory and only re-run them when the underlying data actually changes.

The setup

Your app has a thread-local SQLite connection with row_factory set to sqlite3.Row. The db handle never touches the cache key -- just the user_id and any filter params.

import sqlite3
import threading
from inhouse import inhouse_cache
from inhouse.sqlite import query_store, rows_to_dicts

store = query_store(default_ttl=60)
_local = threading.local()

def get_db() -> sqlite3.Connection:
    if not getattr(_local, "conn", None):
        _local.conn = sqlite3.connect("dashboard.db")
        _local.conn.row_factory = sqlite3.Row
    return _local.conn

@inhouse_cache(store=store)
def load_usage_summary(team_id: int) -> dict | None:
    row = get_db().execute("""
        SELECT
            COUNT(*) as total_requests,
            AVG(response_time) as avg_latency,
            SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) as error_count,
            MAX(timestamp) as last_activity
        FROM requests
        WHERE team_id = ?
    """, (team_id,)).fetchone()
    return rows_to_dicts(row) if row else None

@inhouse_cache(store=store)
def load_recent_activity(team_id: int, limit: int = 25) -> list[dict]:
    rows = get_db().execute("""
        SELECT endpoint, response_time, status, timestamp
        FROM requests
        WHERE team_id = ?
        ORDER BY timestamp DESC
        LIMIT ?
    """, (team_id, limit)).fetchall()
    return rows_to_dicts(rows)

The first time a team's dashboard loads, both queries hit SQLite. Every subsequent load inside the 60-second TTL returns cached results. No query parsing, no memory allocation for Row objects, no I/O.

Busting the cache when data changes

The dashboard has an "ingest new data" button. When the team imports a CSV, we need fresh results for those two queries without touching anyone else's cache.

def ingest_csv(team_id: int, filepath: str):
    # parse and insert rows
    with open(filepath) as f:
        reader = csv.DictReader(f)
        for row in reader:
            get_db().execute("INSERT INTO requests ...", row)
        get_db().commit()
    
    # bust only this team's cached queries
    load_usage_summary.cache_clear()
    load_recent_activity.cache_clear()

Each decorated function exposes cache_clear() which deletes only the keys prefixed with that function's module-qualified name. Other functions, other teams -- untouched.

Why this works better than raw functools.lru_cache

• rows_to_dicts converts sqlite3.Row to plain dicts on read, so callers can't mutate cached rows.
• The db connection is resolved inside the function from thread-local storage, not passed as an argument, so it never pollutes the cache key.
• The 60-second TTL acts as a safety net. If the ingest button breaks, stale results auto-expire within a minute.
• If you need longer TTLs for slower-moving data, just pass a bigger number to query_store(default_ttl=300).

Going further

Add a second store for long-lived reference data that rarely changes:

hot_store = query_store(default_ttl=30)     # dashboard queries
cold_store = query_store(default_ttl=3600)  # lookup tables

@inhouse_cache(store=cold_store)
def load_team_settings(team_id: int) -> dict:
    ...