117 lines
4.2 KiB
Python
117 lines
4.2 KiB
Python
"""Transactional, versioned SQLite schema migrations for Magent."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Callable
|
|
|
|
|
|
MigrationStep = Callable[[sqlite3.Connection], None]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Migration:
|
|
version: int
|
|
name: str
|
|
apply: MigrationStep
|
|
|
|
|
|
def _column_names(conn: sqlite3.Connection, table: str) -> set[str]:
|
|
return {str(row[1]) for row in conn.execute(f'PRAGMA table_info("{table}")').fetchall()}
|
|
|
|
|
|
def _add_column(conn: sqlite3.Connection, table: str, definition: str) -> None:
|
|
column = definition.split(maxsplit=1)[0].strip('"')
|
|
if column not in _column_names(conn, table):
|
|
conn.execute(f'ALTER TABLE "{table}" ADD COLUMN {definition}')
|
|
|
|
|
|
def _migration_001_legacy_columns_and_indexes(conn: sqlite3.Connection) -> None:
|
|
for definition in (
|
|
"email TEXT",
|
|
"last_login_at TEXT",
|
|
"is_blocked INTEGER NOT NULL DEFAULT 0",
|
|
"auth_provider TEXT NOT NULL DEFAULT 'local'",
|
|
"jellyfin_password_hash TEXT",
|
|
"last_jellyfin_auth_at TEXT",
|
|
"jellyseerr_user_id INTEGER",
|
|
"auto_search_enabled INTEGER NOT NULL DEFAULT 1",
|
|
"invite_management_enabled INTEGER NOT NULL DEFAULT 0",
|
|
"profile_id INTEGER",
|
|
"expires_at TEXT",
|
|
"invited_by_code TEXT",
|
|
"invited_at TEXT",
|
|
"auth_version INTEGER NOT NULL DEFAULT 1",
|
|
):
|
|
_add_column(conn, "users", definition)
|
|
|
|
for definition in ("recipient_email TEXT", "code_hint TEXT"):
|
|
_add_column(conn, "signup_invites", definition)
|
|
|
|
for definition in (
|
|
"related_item_id INTEGER",
|
|
"workflow_request_status TEXT",
|
|
"workflow_media_status TEXT",
|
|
"issue_type TEXT",
|
|
"issue_resolved_at TEXT",
|
|
"metadata_json TEXT",
|
|
):
|
|
_add_column(conn, "portal_items", definition)
|
|
|
|
_add_column(conn, "requests_cache", "requested_by_id INTEGER")
|
|
|
|
statements = (
|
|
"CREATE INDEX IF NOT EXISTS idx_portal_items_workflow ON portal_items "
|
|
"(kind, workflow_request_status, workflow_media_status, updated_at DESC, id DESC)",
|
|
"CREATE INDEX IF NOT EXISTS idx_portal_items_related_item ON portal_items "
|
|
"(related_item_id, updated_at DESC, id DESC)",
|
|
"CREATE INDEX IF NOT EXISTS idx_users_profile_id ON users (profile_id)",
|
|
"CREATE INDEX IF NOT EXISTS idx_users_expires_at ON users (expires_at)",
|
|
"CREATE INDEX IF NOT EXISTS idx_users_username_nocase ON users (username COLLATE NOCASE)",
|
|
"CREATE INDEX IF NOT EXISTS idx_users_email_nocase ON users (email COLLATE NOCASE)",
|
|
"CREATE INDEX IF NOT EXISTS idx_requests_cache_requested_by_id ON requests_cache (requested_by_id)",
|
|
"CREATE INDEX IF NOT EXISTS idx_requests_cache_requested_by_id_created_at ON requests_cache "
|
|
"(requested_by_id, created_at DESC, request_id DESC)",
|
|
)
|
|
for statement in statements:
|
|
conn.execute(statement)
|
|
|
|
|
|
MIGRATIONS = (
|
|
Migration(1, "legacy_columns_and_indexes", _migration_001_legacy_columns_and_indexes),
|
|
)
|
|
|
|
|
|
def run_schema_migrations(conn: sqlite3.Connection) -> list[int]:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS schema_migrations (
|
|
version INTEGER PRIMARY KEY,
|
|
name TEXT NOT NULL UNIQUE,
|
|
applied_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
applied = {int(row[0]) for row in conn.execute("SELECT version FROM schema_migrations")}
|
|
completed: list[int] = []
|
|
for migration in MIGRATIONS:
|
|
if migration.version in applied:
|
|
continue
|
|
savepoint = f"magent_migration_{migration.version}"
|
|
conn.execute(f"SAVEPOINT {savepoint}")
|
|
try:
|
|
migration.apply(conn)
|
|
conn.execute(
|
|
"INSERT INTO schema_migrations (version, name, applied_at) VALUES (?, ?, ?)",
|
|
(migration.version, migration.name, datetime.now(timezone.utc).isoformat()),
|
|
)
|
|
conn.execute(f"RELEASE SAVEPOINT {savepoint}")
|
|
except Exception:
|
|
conn.execute(f"ROLLBACK TO SAVEPOINT {savepoint}")
|
|
conn.execute(f"RELEASE SAVEPOINT {savepoint}")
|
|
raise
|
|
completed.append(migration.version)
|
|
return completed
|