57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
"""SQLite persistence for workbench click counts."""
|
|
|
|
import os
|
|
import sqlite3
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
BACKEND_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def database_path() -> Path:
|
|
configured = Path(os.getenv("PORTAL_DATABASE", "./data/usage.sqlite3")).expanduser()
|
|
return configured if configured.is_absolute() else BACKEND_ROOT / configured
|
|
|
|
|
|
def connect(path: Path | None = None) -> sqlite3.Connection:
|
|
resolved = path or database_path()
|
|
resolved.parent.mkdir(parents=True, exist_ok=True)
|
|
connection = sqlite3.connect(resolved, timeout=15)
|
|
connection.row_factory = sqlite3.Row
|
|
connection.execute("PRAGMA busy_timeout=15000")
|
|
connection.execute("PRAGMA foreign_keys=ON")
|
|
return connection
|
|
|
|
|
|
def initialize(path: Path | None = None) -> None:
|
|
with connect(path) as connection:
|
|
connection.execute("PRAGMA journal_mode=WAL")
|
|
connection.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS application_usage (
|
|
app_id TEXT PRIMARY KEY,
|
|
click_count INTEGER NOT NULL DEFAULT 0 CHECK (click_count >= 0),
|
|
updated_at TEXT
|
|
)
|
|
"""
|
|
)
|
|
connection.commit()
|
|
|
|
|
|
@contextmanager
|
|
def session(path: Path | None = None):
|
|
connection = connect(path)
|
|
try:
|
|
yield connection
|
|
connection.commit()
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|