Spaces:
Running
Running
"""SQLite data access layer for MiniWeb — per-site tables.
Each site has its own tables (e.g., banking_users, banking_transactions)
with real columns and indexes. Session mutations are stored in a generic
session_overlay table and merged at query time.
Core API
--------
query(site, collection, **filters) -> list[dict]
get_item(site, collection, item_id) -> dict | None
count(site, collection, **filters) -> int
save_item(site, collection, item_id, data) -> None
delete_item(site, collection, item_id) -> None
Session management
------------------
reset_session(sid) — revert session to pristine state
reset_all() — clear all sessions
get_stats() — overlay statistics
Schema & build
--------------
init_db() — create infrastructure tables
create_site_table() — create a per-site table from schema
get_table_name() — look up (site, collection) -> SQL table name
"""
from helpers.paths import REPO_ROOT
from miniweb.app import uploads
import json
import os
import pathlib
import sqlite3
import threading
import time
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
# MINIWEB_DB, relative to the repository unless absolute; by default data/miniweb.db, where
# scripts/restore_site_db.py writes it
_DB_PATH = str(REPO_ROOT / os.environ.get("MINIWEB_DB", "data/miniweb.db"))
_SESSION_TTL = 3600 # 1 hour
# ---------------------------------------------------------------------------
# Connection pool — one connection per thread (SQLite requirement)
# ---------------------------------------------------------------------------
_local = threading.local()
def _get_conn() -> sqlite3.Connection:
"""Return a thread-local SQLite connection."""
conn = getattr(_local, "conn", None)
if conn is None:
conn = sqlite3.connect(_DB_PATH, timeout=30)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.execute("PRAGMA cache_size=-64000") # 64 MB cache
conn.execute("PRAGMA foreign_keys=OFF")
conn.row_factory = sqlite3.Row
if _guarded is not None:
conn.set_authorizer(_authorize)
_local.conn = conn
return conn
def get_conn() -> sqlite3.Connection:
"""Public access to the thread-local connection."""
return _get_conn()
def close():
"""Close the thread-local connection (call at teardown)."""
conn = getattr(_local, "conn", None)
if conn is not None:
conn.close()
_local.conn = None
# ---------------------------------------------------------------------------
# Write guard (the public demo): the site tables refuse every write
# ---------------------------------------------------------------------------
# tables a request may still write: the visitors' own state, and the registry rows that the sites' setup code
# registers again with the same values
_SESSION_TABLES = {"session_overlay", "session_collection_replaced", "sessions", "event_log", "site_registry",
"sqlite_sequence"}
_DROPS = {sqlite3.SQLITE_DROP_TABLE, sqlite3.SQLITE_DROP_INDEX, sqlite3.SQLITE_DROP_TRIGGER, sqlite3.SQLITE_DROP_VIEW,
sqlite3.SQLITE_DROP_VTABLE, sqlite3.SQLITE_ALTER_TABLE}
_guarded = None # the tables that held rows when the guard went on; None: no guard
def _authorize(action, table, *_):
if action in (sqlite3.SQLITE_INSERT, sqlite3.SQLITE_UPDATE, sqlite3.SQLITE_DELETE) and table in _guarded:
return sqlite3.SQLITE_DENY
if action in _DROPS:
return sqlite3.SQLITE_DENY
return sqlite3.SQLITE_OK
def enable_write_guard():
"""Refuse, on every connection from now on, writes to each table that holds site data (demo mode); sessions keep
writing their changes to session_overlay. A table a site creates empty at runtime stays writable, so its one-time
seed (e.g. the forums' moderators) still runs."""
global _guarded
conn = _get_conn()
names = [r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'")]
_guarded = {n for n in names if n not in _SESSION_TABLES
and conn.execute(f'SELECT 1 FROM "{n}" LIMIT 1').fetchone()}
conn.set_authorizer(_authorize)
# ---------------------------------------------------------------------------
# Infrastructure schema
# ---------------------------------------------------------------------------
_INFRA_SCHEMA = """
CREATE TABLE IF NOT EXISTS site_registry (
site TEXT NOT NULL,
collection TEXT NOT NULL,
table_name TEXT NOT NULL,
pk_column TEXT NOT NULL DEFAULT 'id',
PRIMARY KEY (site, collection)
);
CREATE TABLE IF NOT EXISTS session_overlay (
session_id TEXT NOT NULL,
site TEXT NOT NULL,
collection TEXT NOT NULL,
item_id TEXT NOT NULL,
op TEXT NOT NULL DEFAULT 'upsert',
data TEXT,
PRIMARY KEY (session_id, site, collection, item_id)
);
CREATE TABLE IF NOT EXISTS session_collection_replaced (
session_id TEXT NOT NULL,
site TEXT NOT NULL,
collection TEXT NOT NULL,
PRIMARY KEY (session_id, site, collection)
);
CREATE TABLE IF NOT EXISTS sessions (
session_id TEXT PRIMARY KEY,
last_access TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_overlay_session
ON session_overlay (session_id);
CREATE INDEX IF NOT EXISTS idx_overlay_site_coll
ON session_overlay (site, collection, session_id);
CREATE TABLE IF NOT EXISTS event_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
session_id TEXT NOT NULL,
event_type TEXT NOT NULL,
source_site TEXT NOT NULL DEFAULT '',
payload TEXT NOT NULL DEFAULT '{}',
handlers_called INTEGER NOT NULL DEFAULT 0,
errors TEXT NOT NULL DEFAULT ''
);
CREATE INDEX IF NOT EXISTS idx_event_log_type ON event_log (event_type);
CREATE INDEX IF NOT EXISTS idx_event_log_session ON event_log (session_id);
"""
def init_db(db_path: str = None):
"""Create infrastructure tables. Called once at app startup."""
global _DB_PATH
if db_path:
_DB_PATH = db_path
conn = _get_conn()
conn.executescript(_INFRA_SCHEMA)
conn.commit()
# ---------------------------------------------------------------------------
# Site registry — maps (site, collection) to SQL table names
# ---------------------------------------------------------------------------
# In-memory cache for fast lookups
_registry_cache = {} # (site, collection) -> (table_name, pk_column)
def _load_registry():
"""Load the site registry into memory."""
global _registry_cache
conn = _get_conn()
try:
rows = conn.execute("SELECT site, collection, table_name, pk_column FROM site_registry").fetchall()
_registry_cache = {
(row[0], row[1]): (row[2], row[3]) for row in rows
}
except sqlite3.OperationalError:
_registry_cache = {}
def get_table_name(site: str, collection: str) -> str | None:
"""Look up the SQL table name for a (site, collection) pair."""
if not _registry_cache:
_load_registry()
entry = _registry_cache.get((site, collection))
return entry[0] if entry else None
def get_pk_column(site: str, collection: str) -> str:
"""Get the primary key column name for a (site, collection) pair."""
if not _registry_cache:
_load_registry()
entry = _registry_cache.get((site, collection))
return entry[1] if entry else "id"
def list_collections(site: str) -> list[str]:
"""List all collections for a site."""
if not _registry_cache:
_load_registry()
return [coll for (s, coll) in _registry_cache if s == site]
def register_table(site: str, collection: str, table_name: str, pk_column: str = "id",
conn: sqlite3.Connection = None):
"""Register a site table in the registry."""
if conn is None:
conn = _get_conn()
conn.execute(
"INSERT OR REPLACE INTO site_registry (site, collection, table_name, pk_column) "
"VALUES (?, ?, ?, ?)",
(site, collection, table_name, pk_column),
)
_registry_cache[(site, collection)] = (table_name, pk_column)
# ---------------------------------------------------------------------------
# Session helpers
# ---------------------------------------------------------------------------
def _get_session_id() -> str:
"""Get the current Flask session ID, or '_no_session' outside request."""
try:
from flask import session, has_request_context
if has_request_context():
sid = session.get("_data_overlay_sid")
if not sid:
import uuid
sid = str(uuid.uuid4())[:12]
session["_data_overlay_sid"] = sid
return sid
except (ImportError, RuntimeError):
pass
return "_no_session"
def _touch_session(conn: sqlite3.Connection, sid: str):
"""Update session last_access timestamp."""
now = time.strftime("%Y-%m-%dT%H:%M:%S")
conn.execute(
"INSERT INTO sessions (session_id, last_access) VALUES (?, ?) "
"ON CONFLICT(session_id) DO UPDATE SET last_access = excluded.last_access",
(sid, now),
)
def reset_session(sid: str = None):
"""Clear a session's overlay (revert to pristine)."""
if sid is None:
sid = _get_session_id()
conn = _get_conn()
conn.execute("DELETE FROM session_overlay WHERE session_id = ?", (sid,))
conn.execute("DELETE FROM session_collection_replaced WHERE session_id = ?", (sid,))
conn.execute("DELETE FROM sessions WHERE session_id = ?", (sid,))
conn.commit()
uploads.drop([sid])
def reset_all():
"""Clear all session overlays."""
conn = _get_conn()
conn.execute("DELETE FROM session_overlay")
conn.execute("DELETE FROM session_collection_replaced")
conn.execute("DELETE FROM sessions")
conn.commit()
uploads.drop_all()
def evict_stale_sessions():
"""Remove sessions idle longer than TTL."""
conn = _get_conn()
cutoff = time.strftime(
"%Y-%m-%dT%H:%M:%S",
time.localtime(time.time() - _SESSION_TTL),
)
stale = [
row[0]
for row in conn.execute(
"SELECT session_id FROM sessions WHERE last_access < ?", (cutoff,)
).fetchall()
]
if stale:
placeholders = ",".join("?" * len(stale))
conn.execute(
f"DELETE FROM session_overlay WHERE session_id IN ({placeholders})",
stale,
)
conn.execute(
f"DELETE FROM session_collection_replaced WHERE session_id IN ({placeholders})",
stale,
)
conn.execute(
f"DELETE FROM sessions WHERE session_id IN ({placeholders})",
stale,
)
conn.commit()
uploads.drop(stale)
def get_stats() -> dict:
"""Return overlay statistics for debugging."""
conn = _get_conn()
sessions = conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0]
total_files = conn.execute("SELECT COUNT(*) FROM session_overlay").fetchone()[0]
details = {}
for row in conn.execute(
"SELECT s.session_id, "
" (SELECT COUNT(*) FROM session_overlay o WHERE o.session_id = s.session_id) AS files, "
" s.last_access "
"FROM sessions s"
).fetchall():
details[row[0]] = {"files": row[1], "last_access": row[2]}
return {
"sessions": sessions,
"total_files": total_files,
"session_ttl_seconds": _SESSION_TTL,
"sessions_detail": details,
}
# ---------------------------------------------------------------------------
# Overlay helpers — fetch the (tiny) session mutations
# ---------------------------------------------------------------------------
def _get_overlay(sid: str, site: str, collection: str, conn: sqlite3.Connection):
"""Return (deletes: set[str], upserts: dict[str, dict]) for this session."""
rows = conn.execute(
"SELECT item_id, op, data FROM session_overlay "
"WHERE session_id = ? AND site = ? AND collection = ?",
(sid, site, collection),
).fetchall()
deletes = set()
upserts = {}
for row in rows:
item_id, op, data_str = row[0], row[1], row[2]
if op == "delete":
deletes.add(str(item_id))
else:
upserts[str(item_id)] = json.loads(data_str)
return deletes, upserts
def _is_collection_replaced(sid: str, site: str, collection: str, conn: sqlite3.Connection) -> bool:
"""Check if this session fully replaced the collection."""
return conn.execute(
"SELECT 1 FROM session_collection_replaced "
"WHERE session_id = ? AND site = ? AND collection = ?",
(sid, site, collection),
).fetchone() is not None
# ---------------------------------------------------------------------------
# Query API — the main interface sites use
# ---------------------------------------------------------------------------
def query(
site: str,
collection: str,
*,
where: dict = None,
sort: str = None,
limit: int = None,
offset: int = 0,
sid: str = None,
) -> list[dict]:
"""Query items from a site table with SQL-level filtering.
Args:
site: Site name (e.g., "banking")
collection: Collection name (e.g., "transactions")
where: Filter dict, e.g., {"user_id": 1, "status": "posted"}
sort: Column to sort by. Prefix with "-" for DESC (e.g., "-date")
limit: Max rows to return
offset: Skip this many rows
sid: Session ID (auto-detected if None)
Returns:
List of dicts, one per matching row.
"""
if sid is None:
sid = _get_session_id()
conn = _get_conn()
_touch_session(conn, sid)
table = get_table_name(site, collection)
if table is None:
conn.commit()
return []
pk_col = get_pk_column(site, collection)
replaced = _is_collection_replaced(sid, site, collection, conn)
deletes, upserts = _get_overlay(sid, site, collection, conn)
if replaced:
# Session owns the full collection — return only overlay data
items = list(upserts.values())
items = _apply_filters(items, where)
items = _apply_sort(items, sort)
if offset:
items = items[offset:]
if limit is not None:
items = items[:limit]
conn.commit()
return items
# Build SQL query against the base table
sql = f"SELECT * FROM [{table}]"
params = []
clauses = []
if where:
for col, val in where.items():
clauses.append(f"[{col}] = ?")
params.append(val)
# Exclude deleted items
if deletes:
placeholders = ",".join("?" * len(deletes))
clauses.append(f"[{pk_col}] NOT IN ({placeholders})")
params.extend(deletes)
# Exclude items that have been upserted (we'll add the new version later)
if upserts:
placeholders = ",".join("?" * len(upserts))
clauses.append(f"CAST([{pk_col}] AS TEXT) NOT IN ({placeholders})")
params.extend(upserts.keys())
if clauses:
sql += " WHERE " + " AND ".join(clauses)
if sort:
desc = sort.startswith("-")
col = sort.lstrip("-")
direction = "DESC" if desc else "ASC"
sql += f" ORDER BY [{col}] {direction}"
if limit is not None:
# Fetch extra to account for upserts we'll merge in
fetch_limit = limit + len(upserts) + offset
sql += f" LIMIT {fetch_limit}"
try:
rows = conn.execute(sql, params).fetchall()
except sqlite3.OperationalError:
conn.commit()
return []
results = [_deserialize_row(row) for row in rows]
# Merge upserts that match the filters
if upserts:
matching_upserts = _apply_filters(list(upserts.values()), where)
results.extend(matching_upserts)
results = _apply_sort(results, sort)
# Apply offset and limit after merge
if offset:
results = results[offset:]
if limit is not None:
results = results[:limit]
conn.commit()
return results
def get_item(site: str, collection: str, item_id, sid: str = None) -> dict | None:
"""Fetch a single item by primary key.
Checks session overlay first, then falls back to base table.
"""
if sid is None:
sid = _get_session_id()
conn = _get_conn()
_touch_session(conn, sid)
item_id_str = str(item_id)
# Check overlay first
overlay_row = conn.execute(
"SELECT op, data FROM session_overlay "
"WHERE session_id = ? AND site = ? AND collection = ? AND item_id = ?",
(sid, site, collection, item_id_str),
).fetchone()
if overlay_row:
if overlay_row[0] == "delete":
conn.commit()
return None
conn.commit()
return json.loads(overlay_row[1])
# Check if collection was fully replaced (item not in overlay = doesn't exist)
if _is_collection_replaced(sid, site, collection, conn):
conn.commit()
return None
# Fetch from base table
table = get_table_name(site, collection)
if table is None:
conn.commit()
return None
pk_col = get_pk_column(site, collection)
try:
row = conn.execute(
f"SELECT * FROM [{table}] WHERE [{pk_col}] = ?", (item_id,)
).fetchone()
except sqlite3.OperationalError:
conn.commit()
return None
conn.commit()
return _deserialize_row(row) if row else None
def count(
site: str,
collection: str,
*,
where: dict = None,
sid: str = None,
) -> int:
"""Count matching items (base + overlay)."""
if sid is None:
sid = _get_session_id()
conn = _get_conn()
_touch_session(conn, sid)
table = get_table_name(site, collection)
if table is None:
conn.commit()
return 0
pk_col = get_pk_column(site, collection)
replaced = _is_collection_replaced(sid, site, collection, conn)
deletes, upserts = _get_overlay(sid, site, collection, conn)
if replaced:
items = _apply_filters(list(upserts.values()), where)
conn.commit()
return len(items)
# Count from base table
sql = f"SELECT COUNT(*) FROM [{table}]"
params = []
clauses = []
if where:
for col, val in where.items():
clauses.append(f"[{col}] = ?")
params.append(val)
if deletes:
placeholders = ",".join("?" * len(deletes))
clauses.append(f"[{pk_col}] NOT IN ({placeholders})")
params.extend(deletes)
if upserts:
placeholders = ",".join("?" * len(upserts))
clauses.append(f"CAST([{pk_col}] AS TEXT) NOT IN ({placeholders})")
params.extend(upserts.keys())
if clauses:
sql += " WHERE " + " AND ".join(clauses)
try:
base_count = conn.execute(sql, params).fetchone()[0]
except sqlite3.OperationalError:
base_count = 0
# Add matching upserts
upsert_count = len(_apply_filters(list(upserts.values()), where))
conn.commit()
return base_count + upsert_count
def save_item(site: str, collection: str, item_id, data: dict, sid: str = None):
"""Save (upsert) a single item in the session overlay."""
if sid is None:
sid = _get_session_id()
conn = _get_conn()
_touch_session(conn, sid)
conn.execute(
"INSERT OR REPLACE INTO session_overlay "
"(session_id, site, collection, item_id, op, data) "
"VALUES (?, ?, ?, ?, 'upsert', ?)",
(sid, site, collection, str(item_id), json.dumps(data, ensure_ascii=False)),
)
conn.commit()
def next_id(site: str, collection: str) -> int:
"""Next integer primary key for a collection, aware of BOTH the base table and
THIS session's overlay.
Using only `MAX(id)` from the base table collides on the 2nd create within a
session: the 1st new item lives only in session_overlay, so the base MAX is
unchanged and the same id is handed out again (→ a UNIQUE-constraint crash with
save_collection, or a silent overwrite with save_item). Integer PKs only.
"""
conn = _get_conn()
pk = get_pk_column(site, collection)
table = get_table_name(site, collection)
base = 0
if table:
try:
# CAST so a TEXT pk with non-numeric ids (e.g. 'user_5') yields 0 in SQLite
# instead of returning a string that then crashes int() below.
base = conn.execute(f"SELECT MAX(CAST([{pk}] AS INTEGER)) FROM [{table}]").fetchone()[0] or 0
except sqlite3.OperationalError:
base = 0
over = conn.execute(
"SELECT MAX(CAST(item_id AS INTEGER)) FROM session_overlay "
"WHERE session_id = ? AND site = ? AND collection = ?",
(_get_session_id(), site, collection),
).fetchone()[0] or 0
return max(int(base), int(over)) + 1
def delete_item(site: str, collection: str, item_id, sid: str = None):
"""Mark an item as deleted in the session overlay."""
if sid is None:
sid = _get_session_id()
conn = _get_conn()
_touch_session(conn, sid)
conn.execute(
"INSERT OR REPLACE INTO session_overlay "
"(session_id, site, collection, item_id, op, data) "
"VALUES (?, ?, ?, ?, 'delete', NULL)",
(sid, site, collection, str(item_id)),
)
conn.commit()
def save_collection(site: str, collection: str, items: list, sid: str = None):
"""Replace an entire collection for this session.
The base table is NOT modified. The session is marked as having
fully replaced this collection, and all items go into session_overlay.
"""
if sid is None:
sid = _get_session_id()
conn = _get_conn()
_touch_session(conn, sid)
pk_col = get_pk_column(site, collection)
conn.execute(
"INSERT OR REPLACE INTO session_collection_replaced "
"(session_id, site, collection) VALUES (?, ?, ?)",
(sid, site, collection),
)
conn.execute(
"DELETE FROM session_overlay "
"WHERE session_id = ? AND site = ? AND collection = ?",
(sid, site, collection),
)
conn.executemany(
"INSERT INTO session_overlay "
"(session_id, site, collection, item_id, op, data) "
"VALUES (?, ?, ?, ?, 'upsert', ?)",
[
(sid, site, collection, str(item.get(pk_col, i)),
json.dumps(item, ensure_ascii=False))
for i, item in enumerate(items)
],
)
conn.commit()
def _canon(v):
"""A comparable form of a stored value: base columns come back as text, overlay rows as
JSON types (True vs "1", 3 vs "3.0"), so scalars compare as normalized text."""
if isinstance(v, bool):
return "1" if v else "0"
if v is None:
return ""
if isinstance(v, (int, float)):
return str(int(v)) if float(v).is_integer() else repr(float(v))
if isinstance(v, (list, tuple)):
return [_canon(x) for x in v]
if isinstance(v, dict):
return {k: _canon(x) for k, x in v.items()}
s = str(v)
try:
f = float(s)
return str(int(f)) if f.is_integer() and s.strip().lstrip("-").replace(".", "", 1).isdigit() else s
except ValueError:
return s
def session_changes(sid: str = None, site: str = None, replaced_cap: int = 5000) -> list[dict]:
"""This session's effective changes against the base tables, smallest first:
[{site, collection, item_id, op: insert|update|delete, changed: {field: [before, after]},
after: row|None}]. Overlay rows equal to their base row (an undone toggle, the untouched
rows of a replaced collection) are not changes. For privileged checks (datagen), never
for site code."""
if sid is None:
sid = _get_session_id()
conn = _get_conn()
q = "SELECT site, collection, item_id, op, data FROM session_overlay WHERE session_id = ?"
args = [sid]
if site:
q += " AND site = ?"
args.append(site)
rows = conn.execute(q, args).fetchall()
replaced = {(r[0], r[1]) for r in conn.execute(
"SELECT site, collection FROM session_collection_replaced WHERE session_id = ?", (sid,)).fetchall()
if not site or r[0] == site}
out, seen = [], set()
def base_row(st, coll, item_id):
table = get_table_name(st, coll)
if not table:
return None
pk = get_pk_column(st, coll)
try:
row = conn.execute(f"SELECT * FROM [{table}] WHERE CAST([{pk}] AS TEXT) = ? LIMIT 1", (str(item_id),)).fetchone()
except sqlite3.OperationalError:
return None
return _deserialize_row(row) if row else None
for st, coll, item_id, op, data in rows:
seen.add((st, coll, str(item_id)))
before = base_row(st, coll, item_id)
if op == "delete":
if before is not None:
out.append({"site": st, "collection": coll, "item_id": str(item_id), "op": "delete",
"changed": {}, "after": None})
continue
after = json.loads(data) if data else {}
if before is None:
out.append({"site": st, "collection": coll, "item_id": str(item_id), "op": "insert",
"changed": {k: [None, v] for k, v in after.items()}, "after": after})
continue
changed = {k: [before.get(k), after.get(k)] for k in set(before) | set(after)
if _canon(before.get(k)) != _canon(after.get(k))}
if changed:
out.append({"site": st, "collection": coll, "item_id": str(item_id), "op": "update",
"changed": changed, "after": after})
for st, coll in replaced: # a replaced collection's base rows it dropped were deleted
table = get_table_name(st, coll)
if not table:
continue
pk = get_pk_column(st, coll)
for (item_id,) in conn.execute(f"SELECT CAST([{pk}] AS TEXT) FROM [{table}] LIMIT ?", (replaced_cap,)).fetchall():
if (st, coll, str(item_id)) not in seen:
out.append({"site": st, "collection": coll, "item_id": str(item_id), "op": "delete",
"changed": {}, "after": None})
conn.commit()
return out
# ---------------------------------------------------------------------------
# Raw SQL — for sites that need custom queries
# ---------------------------------------------------------------------------
def search(
site: str,
collection: str,
q: str,
*,
where: dict = None,
limit: int = 50,
offset: int = 0,
) -> list[dict]:
"""Full-text search with BM25 ranking via FTS5.
Falls back to LIKE-based search if FTS5 index doesn't exist.
Args:
site: Site name
collection: Collection name
q: Search query string
where: Additional equality filters
limit: Max results
offset: Skip rows
Returns:
List of dicts ranked by relevance (best first).
"""
if not q or not q.strip():
return query(site, collection, where=where, limit=limit, offset=offset)
conn = _get_conn()
table = get_table_name(site, collection)
if table is None:
return []
pk_col = get_pk_column(site, collection)
fts_table = f"fts_{table}"
# Sanitize query for FTS5: escape special chars, convert to prefix match
terms = q.strip().split()
fts_query = " ".join(f'"{t}"*' for t in terms if t)
# Try FTS5 first
try:
# Check if FTS index is populated; skip to LIKE fallback if empty
fts_count = conn.execute(
f"SELECT COUNT(*) FROM [{fts_table}]"
).fetchone()[0]
if fts_count > 0:
sql_parts = [f"SELECT t.* FROM [{table}] t"]
sql_parts.append(f"JOIN [{fts_table}] fts ON t.[{pk_col}] = fts.rowid")
params = []
where_clauses = [f"[{fts_table}] MATCH ?"]
params.append(fts_query)
if where:
for col, val in where.items():
where_clauses.append(f"t.[{col}] = ?")
params.append(val)
sql_parts.append("WHERE " + " AND ".join(where_clauses))
sql_parts.append("ORDER BY fts.rank")
sql_parts.append(f"LIMIT ? OFFSET ?")
params.extend([limit, offset])
sql = " ".join(sql_parts)
rows = conn.execute(sql, params).fetchall()
return [_deserialize_row(r) for r in rows]
except sqlite3.OperationalError:
pass
# Fallback: LIKE-based search on all TEXT columns
try:
col_info = conn.execute(f"PRAGMA table_info([{table}])").fetchall()
text_cols = [r[1] for r in col_info if "TEXT" in (r[2] or "").upper()]
except sqlite3.OperationalError:
return []
if not text_cols:
return []
like_clauses = " OR ".join(f"[{c}] LIKE ?" for c in text_cols[:6])
like_param = f"%{q}%"
params = [like_param] * min(len(text_cols), 6)
where_sql = f"({like_clauses})"
if where:
for col, val in where.items():
where_sql += f" AND [{col}] = ?"
params.append(val)
sql = f"SELECT * FROM [{table}] WHERE {where_sql} LIMIT ? OFFSET ?"
params.extend([limit, offset])
try:
rows = conn.execute(sql, params).fetchall()
return [_deserialize_row(r) for r in rows]
except sqlite3.OperationalError:
return []
def execute(sql: str, params: tuple = (), fetch: str = "all"):
"""Run arbitrary SQL and return results.
fetch: "all" -> list[dict], "one" -> dict|None, "val" -> scalar
"""
conn = _get_conn()
try:
cursor = conn.execute(sql, params)
except sqlite3.OperationalError:
return [] if fetch == "all" else None
if fetch == "val":
row = cursor.fetchone()
return row[0] if row else None
elif fetch == "one":
row = cursor.fetchone()
return _deserialize_row(row) if row else None
else:
return [_deserialize_row(r) for r in cursor.fetchall()]
def merge_overlay(
site: str,
collection: str,
rows: list,
*,
match=None,
sort: str = None,
limit: int = None,
sid: str = None,
) -> list[dict]:
"""Merge session-overlay mutations into rows fetched via raw execute().
Raw SQL through execute() reads only the base table, so items created or
edited in this session (which live in session_overlay) are invisible to it.
Call this on the result to apply the session's deletes/upserts.
Args:
rows: base-table rows from execute()
match: optional callable(item) -> bool; overlay upserts must pass it
to be merged in (mirror your SQL WHERE clauses here)
sort: db.query()-style sort string, e.g. "-created_at"
limit: max rows after merge
Returns:
rows with this session's deletes removed and matching upserts merged.
"""
if sid is None:
sid = _get_session_id()
conn = _get_conn()
pk_col = get_pk_column(site, collection)
replaced = _is_collection_replaced(sid, site, collection, conn)
deletes, upserts = _get_overlay(sid, site, collection, conn)
if replaced:
# Session owns the full collection — base rows are superseded
results = list(upserts.values())
else:
if not deletes and not upserts:
results = list(rows)
else:
excluded = deletes | set(upserts.keys())
results = [r for r in rows if str(r.get(pk_col)) not in excluded]
results.extend(upserts.values())
if match is not None:
# Only overlay items need the predicate for correctness, but applying
# it uniformly keeps replaced-collection results consistent too.
results = [r for r in results if match(r)]
results = _apply_sort(results, sort)
if limit is not None:
results = results[:limit]
return results
# ---------------------------------------------------------------------------
# Build helpers — used by build_db.py
# ---------------------------------------------------------------------------
def create_site_table(conn: sqlite3.Connection, table_name: str, columns: list,
indexes: list = None):
"""Create a per-site table from a schema definition.
columns: list of (col_name, col_type_with_constraints) tuples
indexes: list of column names or tuples of column names
"""
col_defs = ", ".join(f"[{col}] {ctype}" for col, ctype in columns)
conn.execute(f"CREATE TABLE IF NOT EXISTS [{table_name}] ({col_defs})")
if indexes:
for idx in indexes:
if isinstance(idx, (list, tuple)):
idx_name = f"idx_{table_name}_{'_'.join(idx)}"
cols = ", ".join(f"[{c}]" for c in idx)
else:
idx_name = f"idx_{table_name}_{idx}"
cols = f"[{idx}]"
conn.execute(
f"CREATE INDEX IF NOT EXISTS [{idx_name}] ON [{table_name}] ({cols})"
)
def bulk_insert(conn: sqlite3.Connection, table_name: str, columns: list, records: list):
"""Insert records into a per-site table.
columns: list of (col_name, col_type) tuples (from schema)
records: list of dicts
"""
col_names = [col for col, _ in columns]
placeholders = ", ".join("?" * len(col_names))
col_list = ", ".join(f"[{c}]" for c in col_names)
rows = []
for rec in records:
row = []
for col_name in col_names:
val = rec.get(col_name)
# Serialize lists/dicts to JSON strings
if isinstance(val, (list, dict)):
val = json.dumps(val, ensure_ascii=False)
row.append(val)
rows.append(tuple(row))
conn.executemany(
f"INSERT OR REPLACE INTO [{table_name}] ({col_list}) VALUES ({placeholders})",
rows,
)
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _deserialize_row(row) -> dict:
"""Convert a sqlite3.Row to a dict, auto-parsing JSON string values.
Columns that were stored as TEXT but contain JSON arrays/objects
(e.g., notifications, tags, senses) are deserialized back to Python.
"""
d = dict(row)
for key, val in d.items():
if isinstance(val, str) and len(val) >= 2:
first = val[0]
if (first == '[' and val[-1] == ']') or (first == '{' and val[-1] == '}'):
try:
d[key] = json.loads(val)
except (json.JSONDecodeError, ValueError):
pass
return d
def _apply_filters(items: list, where: dict | None) -> list:
"""Filter a list of dicts by a where clause (Python-side)."""
if not where:
return items
result = []
for item in items:
match = True
for col, val in where.items():
item_val = item.get(col)
if item_val != val and str(item_val) != str(val):
match = False
break
if match:
result.append(item)
return result
def _apply_sort(items: list, sort: str | None) -> list:
"""Sort a list of dicts by a column."""
if not sort or not items:
return items
desc = sort.startswith("-")
col = sort.lstrip("-")
return sorted(items, key=lambda x: (x.get(col) is None, x.get(col)), reverse=desc)
"""SQLite data access layer for MiniWeb — per-site tables.
Each site has its own tables (e.g., banking_users, banking_transactions)
with real columns and indexes. Session mutations are stored in a generic
session_overlay table and merged at query time.
Core API
--------
query(site, collection, **filters) -> list[dict]
get_item(site, collection, item_id) -> dict | None
count(site, collection, **filters) -> int
save_item(site, collection, item_id, data) -> None
delete_item(site, collection, item_id) -> None
Session management
------------------
reset_session(sid) — revert session to pristine state
reset_all() — clear all sessions
get_stats() — overlay statistics
Schema & build
--------------
init_db() — create infrastructure tables
create_site_table() — create a per-site table from schema
get_table_name() — look up (site, collection) -> SQL table name
"""
from helpers.paths import REPO_ROOT
from miniweb.app import uploads
import json
import os
import pathlib
import sqlite3
import threading
import time
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
# MINIWEB_DB, relative to the repository unless absolute; by default data/miniweb.db, where
# scripts/restore_site_db.py writes it
_DB_PATH = str(REPO_ROOT / os.environ.get("MINIWEB_DB", "data/miniweb.db"))
_SESSION_TTL = 3600 # 1 hour
# ---------------------------------------------------------------------------
# Connection pool — one connection per thread (SQLite requirement)
# ---------------------------------------------------------------------------
_local = threading.local()
def _get_conn() -> sqlite3.Connection:
"""Return a thread-local SQLite connection."""
conn = getattr(_local, "conn", None)
if conn is None:
conn = sqlite3.connect(_DB_PATH, timeout=30)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.execute("PRAGMA cache_size=-64000") # 64 MB cache
conn.execute("PRAGMA foreign_keys=OFF")
conn.row_factory = sqlite3.Row
if _guarded is not None:
conn.set_authorizer(_authorize)
_local.conn = conn
return conn
def get_conn() -> sqlite3.Connection:
"""Public access to the thread-local connection."""
return _get_conn()
def close():
"""Close the thread-local connection (call at teardown)."""
conn = getattr(_local, "conn", None)
if conn is not None:
conn.close()
_local.conn = None
# ---------------------------------------------------------------------------
# Write guard (the public demo): the site tables refuse every write
# ---------------------------------------------------------------------------
# tables a request may still write: the visitors' own state, and the registry rows that the sites' setup code
# registers again with the same values
_SESSION_TABLES = {"session_overlay", "session_collection_replaced", "sessions", "event_log", "site_registry",
"sqlite_sequence"}
_DROPS = {sqlite3.SQLITE_DROP_TABLE, sqlite3.SQLITE_DROP_INDEX, sqlite3.SQLITE_DROP_TRIGGER, sqlite3.SQLITE_DROP_VIEW,
sqlite3.SQLITE_DROP_VTABLE, sqlite3.SQLITE_ALTER_TABLE}
_guarded = None # the tables that held rows when the guard went on; None: no guard
def _authorize(action, table, *_):
if action in (sqlite3.SQLITE_INSERT, sqlite3.SQLITE_UPDATE, sqlite3.SQLITE_DELETE) and table in _guarded:
return sqlite3.SQLITE_DENY
if action in _DROPS:
return sqlite3.SQLITE_DENY
return sqlite3.SQLITE_OK