"""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)