"""SQLite database manager for HAD Digital MVP.
Auto-creates schema on first run. Uses WAL mode for concurrent reads.
All tables include created_at and updated_at timestamps.
"""
import sqlite3
import os
import threading
from datetime import datetime, timezone
from pathlib import Path
class Database:
"""Thread-safe SQLite database manager with WAL mode."""
_local = threading.local()
_instance = None
_schema_created = False
def __new__(cls, db_path: str | None = None):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._db_path = db_path
return cls._instance
def _get_connection(self) -> sqlite3.Connection:
"""Get or create a thread-local database connection."""
if not hasattr(self._local, "conn") or self._local.conn is None:
conn = sqlite3.connect(
self._db_path,
detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES,
)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
conn.execute("PRAGMA busy_timeout=5000")
self._local.conn = conn
return self._local.conn
@property
def conn(self) -> sqlite3.Connection:
return self._get_connection()
def initialize(self):
"""Create schema if not exists."""
if self._schema_created:
return
self._create_schema()
self._schema_created = True
def _create_schema(self):
"""Create all database tables."""
conn = self.conn
conn.executescript("""
-- Users table
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL,
salt TEXT NOT NULL,
role TEXT NOT NULL CHECK(role IN (
'oncologist', 'had_nurse', 'community_nurse', 'gp',
'pharmacist', 'patient', 'caregiver', 'admin'
)),
display_name TEXT NOT NULL,
email TEXT,
patient_id INTEGER,
active INTEGER NOT NULL DEFAULT 1,
failed_attempts INTEGER NOT NULL DEFAULT 0,
locked_until TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY (patient_id) REFERENCES patients(id)
);
-- Patients table
CREATE TABLE IF NOT EXISTS patients (
id INTEGER PRIMARY KEY AUTOINCREMENT,
mrn TEXT UNIQUE NOT NULL,
first_name TEXT NOT NULL,
last_name TEXT NOT NULL,
date_of_birth TEXT NOT NULL,
gender TEXT CHECK(gender IN ('M', 'F', 'X')),
phone TEXT,
address TEXT,
emergency_contact TEXT,
primary_oncologist_id INTEGER,
cancer_type TEXT,
cancer_stage TEXT,
diagnosis_date TEXT,
status TEXT NOT NULL DEFAULT 'active' CHECK(status IN ('active', 'discharged', 'deceased')),
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY (primary_oncologist_id) REFERENCES users(id)
);
-- Episodes table
CREATE TABLE IF NOT EXISTS episodes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
patient_id INTEGER NOT NULL,
episode_type TEXT NOT NULL CHECK(episode_type IN (
'chemotherapy', 'radiotherapy', 'immunotherapy',
'targeted_therapy', 'surgery', 'followup'
)),
start_date TEXT NOT NULL,
end_date TEXT,
status TEXT NOT NULL DEFAULT 'active' CHECK(status IN ('active', 'completed', 'cancelled')),
notes TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY (patient_id) REFERENCES patients(id)
);
-- Treatment plans table
CREATE TABLE IF NOT EXISTS treatment_plans (
id INTEGER PRIMARY KEY AUTOINCREMENT,
patient_id INTEGER NOT NULL,
episode_id INTEGER,
protocol_name TEXT NOT NULL,
regimen TEXT NOT NULL,
cycle_count INTEGER,
current_cycle INTEGER DEFAULT 0,
cycle_length_days INTEGER,
drugs TEXT NOT NULL,
start_date TEXT NOT NULL,
end_date TEXT,
status TEXT NOT NULL DEFAULT 'active' CHECK(status IN ('active', 'completed', 'modified', 'discontinued')),
notes TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY (patient_id) REFERENCES patients(id),
FOREIGN KEY (episode_id) REFERENCES episodes(id)
);
-- Toxicity reports table
CREATE TABLE IF NOT EXISTS toxicity_reports (
id INTEGER PRIMARY KEY AUTOINCREMENT,
patient_id INTEGER NOT NULL,
episode_id INTEGER,
treatment_plan_id INTEGER,
reporter_id INTEGER NOT NULL,
symptom_id TEXT NOT NULL,
symptom_category TEXT NOT NULL,
report_date TEXT NOT NULL DEFAULT (datetime('now')),
severity_score REAL,
lab_values TEXT,
notes TEXT,
status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending', 'graded', 'confirmed', 'resolved')),
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY (patient_id) REFERENCES patients(id),
FOREIGN KEY (episode_id) REFERENCES episodes(id),
FOREIGN KEY (treatment_plan_id) REFERENCES treatment_plans(id),
FOREIGN KEY (reporter_id) REFERENCES users(id)
);
-- Toxicity grades table
CREATE TABLE IF NOT EXISTS toxicity_grades (
id INTEGER PRIMARY KEY AUTOINCREMENT,
report_id INTEGER NOT NULL,
patient_id INTEGER NOT NULL,
symptom_id TEXT NOT NULL,
grade INTEGER NOT NULL CHECK(grade BETWEEN 1 AND 5),
criteria TEXT NOT NULL,
intervention TEXT,
provisional INTEGER NOT NULL DEFAULT 1,
confirmed_by INTEGER,
confirmed_at TEXT,
grading_engine_version TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY (report_id) REFERENCES toxicity_reports(id),
FOREIGN KEY (patient_id) REFERENCES patients(id),
FOREIGN KEY (confirmed_by) REFERENCES users(id)
);
-- Alerts table
CREATE TABLE IF NOT EXISTS alerts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
patient_id INTEGER NOT NULL,
grade_id INTEGER,
report_id INTEGER,
alert_type TEXT NOT NULL CHECK(alert_type IN ('routine', 'urgent', 'emergency')),
severity TEXT NOT NULL CHECK(severity IN ('low', 'medium', 'high', 'critical')),
title TEXT NOT NULL,
message TEXT NOT NULL,
assigned_to INTEGER,
assigned_role TEXT,
status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending', 'acknowledged', 'resolved', 'escalated')),
acknowledged_by INTEGER,
acknowledged_at TEXT,
resolved_at TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY (patient_id) REFERENCES patients(id),
FOREIGN KEY (grade_id) REFERENCES toxicity_grades(id),
FOREIGN KEY (report_id) REFERENCES toxicity_reports(id),
FOREIGN KEY (assigned_to) REFERENCES users(id),
FOREIGN KEY (acknowledged_by) REFERENCES users(id)
);
-- Messages table
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sender_id INTEGER NOT NULL,
recipient_id INTEGER,
recipient_role TEXT,
patient_id INTEGER,
subject TEXT,
body TEXT NOT NULL,
message_type TEXT NOT NULL DEFAULT 'direct' CHECK(message_type IN (
'direct', 'alert_notification', 'system', 'care_team'
)),
read INTEGER NOT NULL DEFAULT 0,
read_at TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY (sender_id) REFERENCES users(id),
FOREIGN KEY (recipient_id) REFERENCES users(id),
FOREIGN KEY (patient_id) REFERENCES patients(id)
);
-- Timeline events table
CREATE TABLE IF NOT EXISTS timeline_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
patient_id INTEGER NOT NULL,
episode_id INTEGER,
event_type TEXT NOT NULL CHECK(event_type IN (
'report', 'grade', 'alert', 'message', 'treatment',
'appointment', 'note', 'system'
)),
event_date TEXT NOT NULL DEFAULT (datetime('now')),
title TEXT NOT NULL,
description TEXT,
reference_id INTEGER,
reference_table TEXT,
created_by INTEGER,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY (patient_id) REFERENCES patients(id),
FOREIGN KEY (episode_id) REFERENCES episodes(id),
FOREIGN KEY (created_by) REFERENCES users(id)
);
-- Audit log table (append-only)
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL DEFAULT (datetime('now')),
user_id INTEGER,
username TEXT,
action TEXT NOT NULL,
resource_type TEXT,
resource_id INTEGER,
details TEXT,
ip_address TEXT,
user_agent TEXT,
FOREIGN KEY (user_id) REFERENCES users(id)
);
-- Indexes for performance
CREATE INDEX IF NOT EXISTS idx_patients_mrn ON patients(mrn);
CREATE INDEX IF NOT EXISTS idx_toxicity_reports_patient ON toxicity_reports(patient_id);
CREATE INDEX IF NOT EXISTS idx_toxicity_grades_patient ON toxicity_grades(patient_id);
CREATE INDEX IF NOT EXISTS idx_toxicity_grades_report ON toxicity_grades(report_id);
CREATE INDEX IF NOT EXISTS idx_alerts_patient ON alerts(patient_id);
CREATE INDEX IF NOT EXISTS idx_alerts_status ON alerts(status);
CREATE INDEX IF NOT EXISTS idx_alerts_assigned ON alerts(assigned_to);
CREATE INDEX IF NOT EXISTS idx_messages_recipient ON messages(recipient_id);
CREATE INDEX IF NOT EXISTS idx_messages_patient ON messages(patient_id);
CREATE INDEX IF NOT EXISTS idx_timeline_patient ON timeline_events(patient_id);
CREATE INDEX IF NOT EXISTS idx_timeline_episode ON timeline_events(episode_id);
CREATE INDEX IF NOT EXISTS idx_audit_user ON audit_log(user_id);
CREATE INDEX IF NOT EXISTS idx_audit_timestamp ON audit_log(timestamp);
CREATE INDEX IF NOT EXISTS idx_episodes_patient ON episodes(patient_id);
CREATE INDEX IF NOT EXISTS idx_treatment_plans_patient ON treatment_plans(patient_id);
""")
conn.commit()
def execute(self, sql: str, params: tuple = ()) -> sqlite3.Cursor:
"""Execute a SQL statement."""
return self.conn.execute(sql, params)
def executemany(self, sql: str, params_list: list[tuple]) -> sqlite3.Cursor:
"""Execute a SQL statement with many parameter sets."""
return self.conn.executemany(sql, params_list)
def fetchone(self, sql: str, params: tuple = ()) -> sqlite3.Row | None:
"""Fetch a single row."""
return self.conn.execute(sql, params).fetchone()
def fetchall(self, sql: str, params: tuple = ()) -> list[sqlite3.Row]:
"""Fetch all matching rows."""
return self.conn.execute(sql, params).fetchall()
def commit(self):
"""Commit the current transaction."""
self.conn.commit()
def close(self):
"""Close the thread-local connection."""
if hasattr(self._local, "conn") and self._local.conn is not None:
self._local.conn.close()
self._local.conn = None
def now_iso(self) -> str:
"""Return current UTC time as ISO string."""
return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
# Global singleton
_db_instance = None
def get_db(db_path: str | None = None) -> Database:
"""Get the global database instance."""
global _db_instance
if _db_instance is None:
if db_path is None:
from config_manager import config
db_path = config.get("database", "path", default="data/had.db")
# Ensure directory exists
Path(db_path).parent.mkdir(parents=True, exist_ok=True)
_db_instance = Database(db_path)
_db_instance.initialize()
return _db_instance