Spaces:
Sleeping
Sleeping
File size: 1,963 Bytes
097315f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | """SQLite-backed notebook for saved analyses."""
from __future__ import annotations
import sqlite3
import json
from datetime import datetime
from app.config import NOTEBOOK_DB
def _conn():
NOTEBOOK_DB.parent.mkdir(parents=True, exist_ok=True)
c = sqlite3.connect(str(NOTEBOOK_DB))
c.execute("""
CREATE TABLE IF NOT EXISTS analyses (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT NOT NULL,
title TEXT NOT NULL,
kind TEXT NOT NULL,
wave INTEGER,
spec_json TEXT NOT NULL,
result_json TEXT NOT NULL,
notes TEXT
)
""")
return c
def save_analysis(title: str, kind: str, wave: int, spec: dict, result: dict, notes: str = "") -> int:
c = _conn()
cur = c.execute(
"INSERT INTO analyses (ts, title, kind, wave, spec_json, result_json, notes) VALUES (?,?,?,?,?,?,?)",
(datetime.utcnow().isoformat(), title, kind, wave,
json.dumps(spec, default=str), json.dumps(result, default=str), notes),
)
c.commit()
aid = cur.lastrowid
c.close()
return aid
def list_analyses() -> list[dict]:
c = _conn()
rows = c.execute(
"SELECT id, ts, title, kind, wave, notes FROM analyses ORDER BY ts DESC"
).fetchall()
c.close()
return [
{"id": r[0], "ts": r[1], "title": r[2], "kind": r[3], "wave": r[4], "notes": r[5]}
for r in rows
]
def get_analysis(aid: int) -> dict | None:
c = _conn()
row = c.execute("SELECT * FROM analyses WHERE id = ?", (aid,)).fetchone()
c.close()
if not row:
return None
return {
"id": row[0], "ts": row[1], "title": row[2], "kind": row[3], "wave": row[4],
"spec": json.loads(row[5]), "result": json.loads(row[6]), "notes": row[7],
}
def delete_analysis(aid: int) -> None:
c = _conn()
c.execute("DELETE FROM analyses WHERE id = ?", (aid,))
c.commit()
c.close()
|