300 lines
11 KiB
Python
300 lines
11 KiB
Python
"""Версионированные миграции схемы состояния (issue #20).
|
|
|
|
Схема БД управляется списком :class:`Migration`. Каждая миграция выполняется
|
|
в своей транзакции, запись версии в ``schema_migrations`` — там же: падение
|
|
миграции не оставляет «полуприменённой» схемы.
|
|
|
|
При старте приложения вызывается :func:`migrate` (создание/обновление схемы)
|
|
и :func:`verify_schema` (проверка, что фактическая схема соответствует
|
|
ожидаемой версии).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from dataclasses import dataclass
|
|
|
|
from .clock import utcnow
|
|
from .errors import SchemaError, StorageError
|
|
|
|
SCHEMA_MIGRATIONS = "schema_migrations"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Migration:
|
|
"""Одна версия схемы: имя и список DDL-выражений."""
|
|
|
|
version: int
|
|
name: str
|
|
statements: tuple[str, ...]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class MigrationReport:
|
|
"""Результат :func:`migrate`."""
|
|
|
|
from_version: int
|
|
to_version: int
|
|
applied: tuple[int, ...]
|
|
|
|
@property
|
|
def changed(self) -> bool:
|
|
return bool(self.applied)
|
|
|
|
|
|
_V1_STATEMENTS: tuple[str, ...] = (
|
|
f"""
|
|
CREATE TABLE IF NOT EXISTS {SCHEMA_MIGRATIONS} (
|
|
version INTEGER PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
applied_at TEXT NOT NULL
|
|
)
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS routing_rules (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE,
|
|
kind TEXT NOT NULL CHECK (kind IN ('domain', 'cidr', 'ipset')),
|
|
match TEXT NOT NULL,
|
|
target TEXT NOT NULL,
|
|
priority INTEGER NOT NULL DEFAULT 100,
|
|
enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)),
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS rule_exceptions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
match TEXT NOT NULL UNIQUE,
|
|
reason TEXT NOT NULL DEFAULT '',
|
|
enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)),
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS routing_mode (
|
|
id INTEGER PRIMARY KEY CHECK (id = 1),
|
|
mode TEXT NOT NULL CHECK (mode IN ('off', 'split', 'full')),
|
|
updated_at TEXT NOT NULL
|
|
)
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS geoip_schedule (
|
|
id INTEGER PRIMARY KEY CHECK (id = 1),
|
|
enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)),
|
|
cron TEXT NOT NULL,
|
|
provider TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS configs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE,
|
|
version INTEGER NOT NULL DEFAULT 1,
|
|
path TEXT NOT NULL,
|
|
checksum TEXT NOT NULL DEFAULT '',
|
|
size_bytes INTEGER NOT NULL DEFAULT 0,
|
|
applied INTEGER NOT NULL DEFAULT 0 CHECK (applied IN (0, 1)),
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS devices (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
mac TEXT NOT NULL UNIQUE,
|
|
name TEXT NOT NULL,
|
|
address TEXT NOT NULL DEFAULT '',
|
|
policy TEXT NOT NULL DEFAULT 'default',
|
|
last_seen TEXT,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS audit_log (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
ts TEXT NOT NULL,
|
|
actor TEXT NOT NULL,
|
|
action TEXT NOT NULL,
|
|
entity TEXT NOT NULL,
|
|
entity_id TEXT,
|
|
old_value TEXT,
|
|
new_value TEXT,
|
|
status TEXT NOT NULL DEFAULT 'ok',
|
|
detail TEXT
|
|
)
|
|
""",
|
|
"CREATE INDEX IF NOT EXISTS idx_audit_log_entity ON audit_log (entity, entity_id)",
|
|
"CREATE INDEX IF NOT EXISTS idx_audit_log_ts ON audit_log (ts DESC)",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS state_snapshots (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
entity TEXT NOT NULL,
|
|
entity_id TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
payload TEXT NOT NULL,
|
|
reason TEXT NOT NULL DEFAULT ''
|
|
)
|
|
""",
|
|
"CREATE INDEX IF NOT EXISTS idx_state_snapshots_key ON state_snapshots (entity, entity_id, id)",
|
|
)
|
|
|
|
V1_SEED_STATEMENTS: tuple[str, ...] = (
|
|
"INSERT OR IGNORE INTO routing_mode (id, mode, updated_at) VALUES (1, 'off',"
|
|
" strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
|
|
"INSERT OR IGNORE INTO geoip_schedule (id, enabled, cron, provider, updated_at) VALUES"
|
|
" (1, 1, '0 4 * * *', 'ripe', strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
|
|
)
|
|
|
|
MIGRATIONS: tuple[Migration, ...] = (
|
|
Migration(
|
|
version=1,
|
|
name="initial-state-schema",
|
|
statements=(*_V1_STATEMENTS, *V1_SEED_STATEMENTS),
|
|
),
|
|
)
|
|
|
|
LATEST_VERSION: int = max(migration.version for migration in MIGRATIONS)
|
|
|
|
#: Ожидаемые таблицы схемы v1 и обязательные колонки каждой из них.
|
|
EXPECTED_TABLES: dict[str, frozenset[str]] = {
|
|
SCHEMA_MIGRATIONS: frozenset({"version", "name", "applied_at"}),
|
|
"routing_rules": frozenset(
|
|
{"id", "name", "kind", "match", "target", "priority", "enabled", "created_at", "updated_at"}
|
|
),
|
|
"rule_exceptions": frozenset({"id", "match", "reason", "enabled", "created_at", "updated_at"}),
|
|
"routing_mode": frozenset({"id", "mode", "updated_at"}),
|
|
"geoip_schedule": frozenset({"id", "enabled", "cron", "provider", "updated_at"}),
|
|
"configs": frozenset(
|
|
{
|
|
"id",
|
|
"name",
|
|
"version",
|
|
"path",
|
|
"checksum",
|
|
"size_bytes",
|
|
"applied",
|
|
"created_at",
|
|
"updated_at",
|
|
}
|
|
),
|
|
"devices": frozenset(
|
|
{"id", "mac", "name", "address", "policy", "last_seen", "created_at", "updated_at"}
|
|
),
|
|
"audit_log": frozenset(
|
|
{
|
|
"id",
|
|
"ts",
|
|
"actor",
|
|
"action",
|
|
"entity",
|
|
"entity_id",
|
|
"old_value",
|
|
"new_value",
|
|
"status",
|
|
"detail",
|
|
}
|
|
),
|
|
"state_snapshots": frozenset({"id", "entity", "entity_id", "created_at", "payload", "reason"}),
|
|
}
|
|
|
|
|
|
def table_names(conn: sqlite3.Connection) -> set[str]:
|
|
"""Имена таблиц, фактически присутствующих в БД."""
|
|
rows = conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'").fetchall()
|
|
return {str(row[0]) for row in rows}
|
|
|
|
|
|
def table_columns(conn: sqlite3.Connection, table: str) -> set[str]:
|
|
"""Колонки таблицы (через параметризованное table-valued PRAGMA)."""
|
|
rows = conn.execute("SELECT name FROM pragma_table_info(?)", (table,)).fetchall()
|
|
return {str(row[0]) for row in rows}
|
|
|
|
|
|
def current_version(conn: sqlite3.Connection) -> int:
|
|
"""Текущая версия схемы; ``0`` — если таблицы версий ещё нет."""
|
|
if SCHEMA_MIGRATIONS not in table_names(conn):
|
|
return 0
|
|
row = conn.execute(f"SELECT MAX(version) FROM {SCHEMA_MIGRATIONS}").fetchone() # noqa: S608
|
|
if row is None or row[0] is None:
|
|
return 0
|
|
return int(row[0])
|
|
|
|
|
|
def applied_migrations(conn: sqlite3.Connection) -> dict[int, str]:
|
|
"""Карта «версия -> имя» для применённых миграций."""
|
|
if SCHEMA_MIGRATIONS not in table_names(conn):
|
|
return {}
|
|
rows = conn.execute(f"SELECT version, name FROM {SCHEMA_MIGRATIONS}").fetchall() # noqa: S608
|
|
return {int(row[0]): str(row[1]) for row in rows}
|
|
|
|
|
|
def apply_migration(conn: sqlite3.Connection, migration: Migration) -> None:
|
|
"""Применить одну миграцию в отдельной транзакции (all-or-nothing)."""
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
for statement in migration.statements:
|
|
conn.execute(statement)
|
|
conn.execute(
|
|
f"INSERT INTO {SCHEMA_MIGRATIONS} (version, name, applied_at) VALUES (?, ?, ?)", # noqa: S608
|
|
(migration.version, migration.name, utcnow()),
|
|
)
|
|
except sqlite3.Error as exc: # pragma: no cover - защита от порчи БД
|
|
conn.rollback()
|
|
raise StorageError(
|
|
f"миграция v{migration.version} ({migration.name}) не применена: {exc}"
|
|
) from exc
|
|
except Exception:
|
|
conn.rollback()
|
|
raise
|
|
conn.commit()
|
|
|
|
|
|
def migrate(
|
|
conn: sqlite3.Connection, *, migrations: tuple[Migration, ...] = MIGRATIONS
|
|
) -> MigrationReport:
|
|
"""Применить все неприменённые миграции по возрастанию версии."""
|
|
start = current_version(conn)
|
|
known = sorted(migrations, key=lambda migration: migration.version)
|
|
latest = max((migration.version for migration in known), default=0)
|
|
if start > latest:
|
|
raise SchemaError(
|
|
f"схема БД новее, чем известно приложению: v{start} > v{latest};"
|
|
" требуется обновление образа, не автооткат"
|
|
)
|
|
applied: list[int] = []
|
|
for migration in known:
|
|
if migration.version <= start:
|
|
continue
|
|
apply_migration(conn, migration)
|
|
applied.append(migration.version)
|
|
verify_schema(conn)
|
|
return MigrationReport(
|
|
from_version=start, to_version=current_version(conn), applied=tuple(applied)
|
|
)
|
|
|
|
|
|
def verify_schema(conn: sqlite3.Connection) -> None:
|
|
"""Проверить, что фактическая схема соответствует ожидаемой версии.
|
|
|
|
Вызывает :class:`SchemaError`, если версия схемы новее ожидаемой или если
|
|
отсутствует таблица/колонка, на которую рассчитывает приложение.
|
|
"""
|
|
version = current_version(conn)
|
|
if version > LATEST_VERSION:
|
|
raise SchemaError(f"схема БД v{version} новее ожидаемой v{LATEST_VERSION}")
|
|
if version < LATEST_VERSION:
|
|
raise SchemaError(f"схема БД v{version} не мигрирована до v{LATEST_VERSION}")
|
|
existing = table_names(conn)
|
|
missing_tables = sorted(set(EXPECTED_TABLES) - existing)
|
|
if missing_tables:
|
|
raise SchemaError(f"в схеме отсутствуют таблицы: {', '.join(missing_tables)}")
|
|
for table, required in EXPECTED_TABLES.items():
|
|
missing_columns = sorted(required - table_columns(conn, table))
|
|
if missing_columns:
|
|
raise SchemaError(f"таблица {table}: отсутствуют колонки {', '.join(missing_columns)}")
|