import sqlite3 from collections.abc import Iterator from contextlib import contextmanager from pathlib import Path from fastapi import Request SCHEMA_VERSION = 1 SCHEMA = """ CREATE TABLE IF NOT EXISTS tenants ( id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE COLLATE NOCASE ); -- A job is identified by (tenant, source system, the source's own job id), so two -- products can't collide and renaming a job in Veeam doesn't create a duplicate. CREATE TABLE IF NOT EXISTS jobs ( id INTEGER PRIMARY KEY, tenant_id INTEGER NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, source_system TEXT NOT NULL, external_id TEXT NOT NULL, name TEXT NOT NULL, next_run_at TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, UNIQUE (tenant_id, source_system, external_id) ); CREATE TABLE IF NOT EXISTS runs ( id INTEGER PRIMARY KEY, job_id INTEGER NOT NULL REFERENCES jobs(id) ON DELETE CASCADE, external_id TEXT NOT NULL, status TEXT NOT NULL CHECK (status IN ('running', 'success', 'warning', 'failed')), started_at TEXT NOT NULL, finished_at TEXT, message TEXT NOT NULL DEFAULT '', size_bytes INTEGER, details TEXT NOT NULL DEFAULT '{}', received_at TEXT NOT NULL, CHECK ((status = 'running') = (finished_at IS NULL)), UNIQUE (job_id, external_id) ); CREATE INDEX IF NOT EXISTS runs_job_started ON runs (job_id, started_at); CREATE INDEX IF NOT EXISTS runs_job_finished ON runs (job_id, finished_at); -- Keys used by collectors to push results. Only the SHA-256 is stored. CREATE TABLE IF NOT EXISTS api_keys ( id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE, key_hash TEXT NOT NULL UNIQUE, created_at TEXT NOT NULL, last_used_at TEXT, revoked_at TEXT ); """ def connect(path: Path | str) -> sqlite3.Connection: # Autocommit mode; writes that must be atomic go through transaction(). # check_same_thread=False: FastAPI may run a dependency and its endpoint on different # threadpool threads, but a connection is only ever used by one request at a time. conn = sqlite3.connect(path, timeout=10, isolation_level=None, check_same_thread=False) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") conn.execute("PRAGMA busy_timeout = 10000") return conn def init_db(path: Path | str) -> None: conn = connect(path) try: conn.execute("PRAGMA journal_mode = WAL") version = conn.execute("PRAGMA user_version").fetchone()[0] if version > SCHEMA_VERSION: raise RuntimeError(f"Database schema v{version} is newer than this app (v{SCHEMA_VERSION})") conn.executescript(SCHEMA) conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}") finally: conn.close() @contextmanager def transaction(conn: sqlite3.Connection) -> Iterator[None]: conn.execute("BEGIN IMMEDIATE") try: yield except BaseException: conn.execute("ROLLBACK") raise conn.execute("COMMIT") def get_db(request: Request) -> Iterator[sqlite3.Connection]: conn = connect(request.app.state.settings.db_path) try: yield conn finally: conn.close()