"""reports_store.py — SQLite store for Report Builder templates (DVI-1255). Part of the DVI-1252 Report Builder. A *report template* is a reusable, versioned definition — a set of dashboard-shaped tiles (the chart / report_table tiles registered by P1, DVI-1254) plus a report-level date-range parameter and an optional recipient list — that P3 will render into a viewable report and (later) snapshot to history. Flask-independent (mirrors scada_store.py / idencia_store.py / assets_store.py) so app.py, the togen-worker, and tests can all import it. The DB lives alongside the app's other state files (reports.db next to dashboards.json). Decision D7-A: this lives in a standalone module (NOT inline in app.py) with a DVI-848 dual-dir `update.sh` copy step, so the `sys.path` shadow gotcha can never serve a stale copy. Schema ------ templates(id, name, owner, definition, version, created_at, updated_at, deleted) — one row per template. `definition` is the JSON authoring payload (tiles + date_range + recipients). `version` bumps on every content save; soft-delete via `deleted` keeps ids stable for any P3 run rows that reference them. template_versions(id, template_id, version, definition, created_at, created_by) — append-only history of every saved version, so a template's edit trail (and a future "restore this version") survives overwrites. runs / snapshots — created empty now so P3 (Report Viewer + history + snapshots) is additive with no schema migration; no write API here yet. """ import json import os import sqlite3 import threading import uuid from datetime import datetime from pathlib import Path REPORTS_DB_FILE = Path( os.environ.get("REPORTS_DB_FILE", Path(__file__).resolve().parent / "reports.db")) # Serialize writes within one process; SQLite handles cross-process locking. _write_lock = threading.Lock() def _connect(): conn = sqlite3.connect(str(REPORTS_DB_FILE), timeout=30) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=10000") conn.row_factory = sqlite3.Row return conn def init_reports_db(): """Create tables/indexes if missing. Idempotent; call at import/startup.""" with _write_lock, _connect() as conn: conn.execute( """CREATE TABLE IF NOT EXISTS templates ( id TEXT PRIMARY KEY, name TEXT NOT NULL, owner TEXT NOT NULL DEFAULT '', definition TEXT NOT NULL, version INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, deleted INTEGER NOT NULL DEFAULT 0 )""") conn.execute( """CREATE TABLE IF NOT EXISTS template_versions ( id INTEGER PRIMARY KEY AUTOINCREMENT, template_id TEXT NOT NULL, version INTEGER NOT NULL, definition TEXT NOT NULL, created_at TEXT NOT NULL, created_by TEXT NOT NULL DEFAULT '' )""") conn.execute( """CREATE INDEX IF NOT EXISTS idx_tversions_template ON template_versions(template_id, version)""") # P3 forward-compat: report runs + their data snapshots. Created now so # P3 lands as pure additive code (no migration). No write API yet. conn.execute( """CREATE TABLE IF NOT EXISTS runs ( id TEXT PRIMARY KEY, template_id TEXT NOT NULL, template_version INTEGER, ran_at TEXT NOT NULL, ran_by TEXT NOT NULL DEFAULT '', params TEXT, status TEXT )""") conn.execute( """CREATE TABLE IF NOT EXISTS snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT NOT NULL, tile_id TEXT, payload TEXT NOT NULL )""") conn.execute( """CREATE INDEX IF NOT EXISTS idx_snapshots_run ON snapshots(run_id)""") # P5 (DVI-1258): scheduled-run log. One row per scheduled fire — # doubles as the per-period dedup guard (a fire is skipped when a row # already exists for today's run_date) and the run-health surface the # Report Builder shows. Kept separate from `runs` because an email-only # health record still matters even though every fire also snapshots a # run (email links to that saved run). conn.execute( """CREATE TABLE IF NOT EXISTS schedule_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, template_id TEXT NOT NULL, run_date TEXT NOT NULL, fired_at TEXT NOT NULL, run_id TEXT, emailed INTEGER NOT NULL DEFAULT 0, recipients TEXT, tile_count INTEGER, status TEXT NOT NULL DEFAULT 'ok', error TEXT )""") conn.execute( """CREATE INDEX IF NOT EXISTS idx_schedule_runs_template ON schedule_runs(template_id, run_date)""") def _now(): return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def _row_to_template(row, with_definition=True): if row is None: return None out = { "id": row["id"], "name": row["name"], "owner": row["owner"], "version": row["version"], "created_at": row["created_at"], "updated_at": row["updated_at"], "deleted": bool(row["deleted"]), } if with_definition: try: out["definition"] = json.loads(row["definition"]) except (json.JSONDecodeError, TypeError): out["definition"] = {} return out # --- templates --------------------------------------------------------------- def create_template(name, owner, definition): """Insert a new template at version 1. Returns the full record.""" tid = "rpt_" + uuid.uuid4().hex[:10] now = _now() payload = json.dumps(definition or {}) with _write_lock, _connect() as conn: conn.execute( """INSERT INTO templates (id, name, owner, definition, version, created_at, updated_at, deleted) VALUES (?, ?, ?, ?, 1, ?, ?, 0)""", (tid, name, owner or "", payload, now, now)) conn.execute( """INSERT INTO template_versions (template_id, version, definition, created_at, created_by) VALUES (?, 1, ?, ?, ?)""", (tid, payload, now, owner or "")) return get_template(tid) def get_template(tid, include_deleted=False): """Full template record (incl. definition) or None.""" q = "SELECT * FROM templates WHERE id=?" if not include_deleted: q += " AND deleted=0" with _connect() as conn: row = conn.execute(q, (tid,)).fetchone() return _row_to_template(row) def list_templates(owner=None, include_deleted=False): """Templates (metadata only, no definition) newest-updated first. `owner` filters to one owner's templates (lower-cased match).""" q = "SELECT * FROM templates" conds, args = [], [] if not include_deleted: conds.append("deleted=0") if owner is not None: conds.append("LOWER(owner)=?") args.append(str(owner).strip().lower()) if conds: q += " WHERE " + " AND ".join(conds) q += " ORDER BY updated_at DESC, name COLLATE NOCASE ASC" with _connect() as conn: rows = conn.execute(q, args).fetchall() return [_row_to_template(r, with_definition=False) for r in rows] def update_template(tid, name=None, definition=None, actor=""): """Partial update. A definition change bumps `version` and appends a template_versions row; a name-only change touches `updated_at` without a version bump. Returns the updated record or None if missing/deleted.""" with _write_lock, _connect() as conn: row = conn.execute( "SELECT * FROM templates WHERE id=? AND deleted=0", (tid,)).fetchone() if row is None: return None now = _now() new_name = row["name"] if name is None else name version = row["version"] if definition is not None: payload = json.dumps(definition or {}) version = row["version"] + 1 conn.execute( """UPDATE templates SET name=?, definition=?, version=?, updated_at=? WHERE id=?""", (new_name, payload, version, now, tid)) conn.execute( """INSERT INTO template_versions (template_id, version, definition, created_at, created_by) VALUES (?, ?, ?, ?, ?)""", (tid, version, payload, now, actor or "")) else: conn.execute( "UPDATE templates SET name=?, updated_at=? WHERE id=?", (new_name, now, tid)) return get_template(tid) def delete_template(tid): """Soft-delete (keeps id stable for any P3 run rows). Returns True if a live template was found and marked deleted.""" with _write_lock, _connect() as conn: cur = conn.execute( "UPDATE templates SET deleted=1, updated_at=? WHERE id=? " "AND deleted=0", (_now(), tid)) return cur.rowcount > 0 def template_versions(tid): """Version history (metadata) for one template, newest first.""" with _connect() as conn: rows = conn.execute( "SELECT version, created_at, created_by FROM template_versions " "WHERE template_id=? ORDER BY version DESC", (tid,)).fetchall() return [{"version": r["version"], "created_at": r["created_at"], "created_by": r["created_by"]} for r in rows] def template_count(include_deleted=False): q = "SELECT COUNT(*) AS n FROM templates" if not include_deleted: q += " WHERE deleted=0" with _connect() as conn: return conn.execute(q).fetchone()["n"] # --- runs / snapshots (DVI-1256, DVI-1252 P3) -------------------------------- # A *run* is one execution of a template. Its resolved view-model (the tile # layout + each tile's data payload, already gated per source) is snapshotted # verbatim (decision D4-A) so reopening a historical run reproduces exactly # what it showed and is immune to later data changes. The whole view-model is # stored as ONE snapshots row (tile_id NULL) — the faithful "what it showed" # record — rather than reassembled from per-tile rows at read time. def create_run(template_id, template_version, ran_by, view_model, date_range=None, status="ok"): """Persist a run + its view-model snapshot. Returns the full run record.""" rid = "run_" + uuid.uuid4().hex[:12] now = _now() params = json.dumps({"date_range": date_range or {}}) payload = json.dumps(view_model or {}) with _write_lock, _connect() as conn: conn.execute( """INSERT INTO runs (id, template_id, template_version, ran_at, ran_by, params, status) VALUES (?, ?, ?, ?, ?, ?, ?)""", (rid, template_id, template_version, now, ran_by or "", params, status or "ok")) conn.execute( "INSERT INTO snapshots (run_id, tile_id, payload) VALUES (?, NULL, ?)", (rid, payload)) return get_run(rid) def _row_to_run(row, with_view_model=False, view_model=None): if row is None: return None try: params = json.loads(row["params"]) if row["params"] else {} except (json.JSONDecodeError, TypeError): params = {} out = { "id": row["id"], "template_id": row["template_id"], "template_version": row["template_version"], "ran_at": row["ran_at"], "ran_by": row["ran_by"], "status": row["status"], "date_range": params.get("date_range") or {}, } if with_view_model: out["view_model"] = view_model return out def get_run(rid): """Full run record including its stored view-model, or None.""" with _connect() as conn: row = conn.execute("SELECT * FROM runs WHERE id=?", (rid,)).fetchone() if row is None: return None snap = conn.execute( "SELECT payload FROM snapshots WHERE run_id=? ORDER BY id ASC " "LIMIT 1", (rid,)).fetchone() vm = None if snap is not None: try: vm = json.loads(snap["payload"]) except (json.JSONDecodeError, TypeError): vm = None return _row_to_run(row, with_view_model=True, view_model=vm) def list_runs(template_id, limit=100): """Run history (metadata only, no view-model) newest-run first.""" try: limit = max(1, min(int(limit), 500)) except (TypeError, ValueError): limit = 100 with _connect() as conn: rows = conn.execute( "SELECT * FROM runs WHERE template_id=? " "ORDER BY ran_at DESC, id DESC LIMIT ?", (template_id, limit)).fetchall() return [_row_to_run(r) for r in rows] def run_count(template_id=None): q = "SELECT COUNT(*) AS n FROM runs" args = [] if template_id is not None: q += " WHERE template_id=?" args.append(template_id) with _connect() as conn: return conn.execute(q, args).fetchone()["n"] # --- scheduled-run log (DVI-1258, DVI-1252 P5) ------------------------------- # The scheduler (togen-worker) records every fire here. `run_date` (a local # YYYY-MM-DD) is the per-period dedup key: a daily schedule fires once per day, # and weekly/monthly schedules only reach the runner on their matching day, so # one row per run_date correctly guards all three cadences against a re-fire. def record_schedule_run(template_id, run_date, *, fired_at=None, run_id=None, emailed=False, recipients=None, tile_count=0, status="ok", error=""): """Append a scheduled-run record. Returns the new record's id.""" with _write_lock, _connect() as conn: cur = conn.execute( """INSERT INTO schedule_runs (template_id, run_date, fired_at, run_id, emailed, recipients, tile_count, status, error) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", (template_id, run_date, fired_at or _now(), run_id, 1 if emailed else 0, json.dumps(list(recipients or [])), int(tile_count or 0), status or "ok", (error or "")[:1000])) return cur.lastrowid def _row_to_schedule_run(row): if row is None: return None try: recips = json.loads(row["recipients"]) if row["recipients"] else [] except (json.JSONDecodeError, TypeError): recips = [] return { "id": row["id"], "template_id": row["template_id"], "run_date": row["run_date"], "fired_at": row["fired_at"], "run_id": row["run_id"], "emailed": bool(row["emailed"]), "recipients": recips, "tile_count": row["tile_count"], "status": row["status"], "error": row["error"] or "", } def schedule_ran_on(template_id, run_date): """True if a scheduled run already landed for this template on run_date (the per-period fire guard). A prior error also counts — a failed fire is retried on the next due day, not re-hammered every minute.""" with _connect() as conn: row = conn.execute( "SELECT 1 FROM schedule_runs WHERE template_id=? AND run_date=? " "LIMIT 1", (template_id, run_date)).fetchone() return row is not None def last_schedule_run(template_id): """Most-recent scheduled-run record for a template, or None.""" with _connect() as conn: row = conn.execute( "SELECT * FROM schedule_runs WHERE template_id=? " "ORDER BY id DESC LIMIT 1", (template_id,)).fetchone() return _row_to_schedule_run(row) def list_schedule_runs(template_id, limit=50): """Scheduled-run history (newest first) for the Report Builder health view.""" try: limit = max(1, min(int(limit), 500)) except (TypeError, ValueError): limit = 50 with _connect() as conn: rows = conn.execute( "SELECT * FROM schedule_runs WHERE template_id=? " "ORDER BY id DESC LIMIT ?", (template_id, limit)).fetchall() return [_row_to_schedule_run(r) for r in rows] init_reports_db()