88 lines
2.5 KiB
Python
88 lines
2.5 KiB
Python
"""SQLite helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
|
|
import aiosqlite
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS quizzes (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
title TEXT NOT NULL,
|
|
pool_json TEXT NOT NULL,
|
|
time_limit_default INTEGER NOT NULL DEFAULT 60,
|
|
score_fn_name TEXT NOT NULL DEFAULT 'linear_decay',
|
|
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS quiz_sessions (
|
|
sid TEXT PRIMARY KEY,
|
|
quiz_id INTEGER NOT NULL REFERENCES quizzes(id),
|
|
title TEXT NOT NULL,
|
|
state TEXT NOT NULL DEFAULT 'lobby',
|
|
current_question_idx INTEGER,
|
|
started_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
finished_at TIMESTAMP
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS participants (
|
|
sid TEXT NOT NULL REFERENCES quiz_sessions(sid),
|
|
student_id TEXT NOT NULL,
|
|
name TEXT NOT NULL,
|
|
cookie_id TEXT NOT NULL,
|
|
joined_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
PRIMARY KEY (sid, student_id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS question_events (
|
|
sid TEXT NOT NULL REFERENCES quiz_sessions(sid),
|
|
question_idx INTEGER NOT NULL,
|
|
opened_at TIMESTAMP NOT NULL,
|
|
closed_at TIMESTAMP,
|
|
time_limit INTEGER NOT NULL,
|
|
PRIMARY KEY (sid, question_idx)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS submissions (
|
|
sid TEXT NOT NULL REFERENCES quiz_sessions(sid),
|
|
student_id TEXT NOT NULL,
|
|
question_idx INTEGER NOT NULL,
|
|
answer TEXT,
|
|
submitted_at TIMESTAMP,
|
|
elapsed_ms INTEGER,
|
|
score INTEGER NOT NULL DEFAULT 0,
|
|
status TEXT NOT NULL DEFAULT 'submitted',
|
|
PRIMARY KEY (sid, student_id, question_idx)
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_submissions_sid_qidx ON submissions(sid, question_idx);
|
|
CREATE INDEX IF NOT EXISTS idx_participants_sid ON participants(sid);
|
|
"""
|
|
|
|
|
|
async def init_db(db_path: str) -> None:
|
|
if db_path != ":memory:":
|
|
Path(db_path).parent.mkdir(parents=True, exist_ok=True)
|
|
async with connect(db_path) as db:
|
|
await db.executescript(SCHEMA)
|
|
await db.commit()
|
|
|
|
|
|
@asynccontextmanager
|
|
async def connect(db_path: str) -> AsyncIterator[aiosqlite.Connection]:
|
|
db = await aiosqlite.connect(db_path)
|
|
db.row_factory = aiosqlite.Row
|
|
await db.execute("PRAGMA foreign_keys=ON")
|
|
await db.execute("PRAGMA journal_mode=WAL")
|
|
try:
|
|
yield db
|
|
finally:
|
|
await db.close()
|
|
|
|
|
|
def row_to_dict(row: aiosqlite.Row | None) -> dict | None:
|
|
return dict(row) if row is not None else None
|