From 882ae97fcf63ad5a93b5a8ef051a321704392abe Mon Sep 17 00:00:00 2001 From: Sergey Antropoff Date: Sun, 12 Jul 2026 23:53:26 +0300 Subject: [PATCH] feat: add checksummed database migrations --- Makefile | 2 +- README.md | 5 ++ app/db/migrate_cli.py | 16 +++++ app/db/migrations.py | 70 +++++++++++++++++++++ app/db/migrations/001_initial.sql | 60 ++++++++++++++++++ app/db/primitives.py | 94 ++++++++++++++++++++++++++++ tests/integration/__init__.py | 1 + tests/integration/test_migrations.py | 36 +++++++++++ tests/unit/test_db_primitives.py | 59 +++++++++++++++++ tests/unit/test_migrations.py | 32 ++++++++++ 10 files changed, 374 insertions(+), 1 deletion(-) create mode 100644 app/db/migrate_cli.py create mode 100644 app/db/migrations.py create mode 100644 app/db/migrations/001_initial.sql create mode 100644 app/db/primitives.py create mode 100644 tests/integration/__init__.py create mode 100644 tests/integration/test_migrations.py create mode 100644 tests/unit/test_db_primitives.py create mode 100644 tests/unit/test_migrations.py diff --git a/Makefile b/Makefile index 0fb4f74..cb854c2 100644 --- a/Makefile +++ b/Makefile @@ -63,7 +63,7 @@ db-down: ## Stop PostgreSQL $(COMPOSE) stop postgres db-migrate: ## Apply database migrations - @echo "Database migrations are scheduled for milestone D1" >&2; exit 2 + $(BIN)/python -m app.db.migrate_cli db-reset: ## Recreate the local database volume $(COMPOSE) down -v diff --git a/README.md b/README.md index bf6f091..7349905 100644 --- a/README.md +++ b/README.md @@ -24,8 +24,13 @@ cp .env.example .env make docker-up curl http://localhost:8006/health/live curl http://localhost:8006/health/ready +make db-migrate ``` +Database migrations are ordered SQL files applied transactionally and recorded +with SHA-256 checksums. Re-running `make db-migrate` is safe; changing an already +applied migration is rejected instead of silently drifting the schema. + Contract artifacts can be validated and imported into immutable local revisions: ```bash diff --git a/app/db/migrate_cli.py b/app/db/migrate_cli.py new file mode 100644 index 0000000..9af318f --- /dev/null +++ b/app/db/migrate_cli.py @@ -0,0 +1,16 @@ +"""Apply configured database migrations.""" + +import asyncio + +from app.config import get_settings +from app.db.migrations import migrate_url + + +async def run() -> None: + settings = get_settings() + count = await migrate_url(settings.database_url.get_secret_value()) + print(f"applied {count} migration(s)") + + +if __name__ == "__main__": + asyncio.run(run()) diff --git a/app/db/migrations.py b/app/db/migrations.py new file mode 100644 index 0000000..4c450f1 --- /dev/null +++ b/app/db/migrations.py @@ -0,0 +1,70 @@ +"""Checksummed asynchronous PostgreSQL migration runner.""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from pathlib import Path + +import asyncpg # type: ignore[import-untyped] +from asyncpg import Connection + + +class MigrationError(RuntimeError): + pass + + +@dataclass(frozen=True, slots=True) +class Migration: + version: int + name: str + sql: str + checksum: str + + +def load_migrations(root: Path | None = None) -> tuple[Migration, ...]: + directory = root or Path(__file__).with_name("migrations") + migrations = [] + for path in sorted(directory.glob("[0-9][0-9][0-9]_*.sql")): + version = int(path.name.split("_", 1)[0]) + sql = path.read_text() + migrations.append( + Migration(version, path.stem, sql, hashlib.sha256(sql.encode()).hexdigest()) + ) + return tuple(migrations) + + +async def migrate(connection: Connection, migrations: tuple[Migration, ...] | None = None) -> int: + await connection.execute( + """CREATE TABLE IF NOT EXISTS schema_migrations ( + version integer PRIMARY KEY, name text NOT NULL, checksum text NOT NULL, + applied_at timestamptz NOT NULL DEFAULT now())""" + ) + applied = { + int(row["version"]): str(row["checksum"]) + for row in await connection.fetch("SELECT version, checksum FROM schema_migrations") + } + count = 0 + for migration in migrations or load_migrations(): + if migration.version in applied: + if applied[migration.version] != migration.checksum: + raise MigrationError(f"migration {migration.version} checksum mismatch") + continue + async with connection.transaction(): + await connection.execute(migration.sql) + await connection.execute( + "INSERT INTO schema_migrations(version, name, checksum) VALUES($1, $2, $3)", + migration.version, + migration.name, + migration.checksum, + ) + count += 1 + return count + + +async def migrate_url(database_url: str) -> int: + connection = await asyncpg.connect(database_url) + try: + return await migrate(connection) + finally: + await connection.close() diff --git a/app/db/migrations/001_initial.sql b/app/db/migrations/001_initial.sql new file mode 100644 index 0000000..9a7843f --- /dev/null +++ b/app/db/migrations/001_initial.sql @@ -0,0 +1,60 @@ +CREATE TABLE contract_snapshots ( + checksum text PRIMARY KEY, + source_version text NOT NULL, + document jsonb NOT NULL, + created_at timestamptz NOT NULL DEFAULT now() +); +CREATE TABLE nodes ( + id uuid PRIMARY KEY, + name text NOT NULL UNIQUE, + status text NOT NULL CHECK (status IN ('online', 'offline')) +); +CREATE TABLE resources ( + id uuid PRIMARY KEY, + node_id uuid NOT NULL REFERENCES nodes(id) ON DELETE RESTRICT, + kind text NOT NULL, + external_id text NOT NULL, + state jsonb NOT NULL DEFAULT '{}'::jsonb, + UNIQUE (kind, external_id) +); +CREATE INDEX resources_node_id_idx ON resources(node_id); +CREATE TABLE principals ( + id uuid PRIMARY KEY, + name text NOT NULL UNIQUE, + password_hash text +); +CREATE TABLE roles ( + name text PRIMARY KEY, + privileges text[] NOT NULL DEFAULT '{}' +); +CREATE TABLE acl_entries ( + principal_id uuid NOT NULL REFERENCES principals(id) ON DELETE CASCADE, + role_name text NOT NULL REFERENCES roles(name) ON DELETE RESTRICT, + path text NOT NULL, + propagate boolean NOT NULL DEFAULT true, + PRIMARY KEY (principal_id, role_name, path) +); +CREATE TABLE tasks ( + id uuid PRIMARY KEY, + upid text NOT NULL UNIQUE, + status text NOT NULL, + payload jsonb NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() +); +CREATE INDEX tasks_status_created_idx ON tasks(status, created_at); +CREATE TABLE scenarios ( + id uuid PRIMARY KEY, + name text NOT NULL UNIQUE, + definition jsonb NOT NULL, + enabled boolean NOT NULL DEFAULT true +); +CREATE TABLE audit_events ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + occurred_at timestamptz NOT NULL DEFAULT now(), + principal text, + action text NOT NULL, + target text, + details jsonb NOT NULL DEFAULT '{}'::jsonb +); +CREATE INDEX audit_events_occurred_idx ON audit_events(occurred_at); diff --git a/app/db/primitives.py b/app/db/primitives.py new file mode 100644 index 0000000..e1ec90f --- /dev/null +++ b/app/db/primitives.py @@ -0,0 +1,94 @@ +"""Typed transactional helpers and stable database error mapping.""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator, Awaitable, Callable +from contextlib import asynccontextmanager +from dataclasses import dataclass + +import asyncpg # type: ignore[import-untyped] +from asyncpg import Connection, Pool + + +class DatabaseOperationError(RuntimeError): + """Safe base error for repository operations.""" + + +class ConflictError(DatabaseOperationError): + pass + + +class ReferenceError(DatabaseOperationError): + pass + + +class TransientDatabaseError(DatabaseOperationError): + pass + + +def map_database_error(error: asyncpg.PostgresError) -> DatabaseOperationError: + if isinstance(error, asyncpg.UniqueViolationError): + return ConflictError("database uniqueness constraint failed") + if isinstance(error, asyncpg.ForeignKeyViolationError): + return ReferenceError("database reference constraint failed") + if isinstance( + error, + asyncpg.SerializationError + | asyncpg.DeadlockDetectedError + | asyncpg.TooManyConnectionsError, + ): + return TransientDatabaseError("transient database failure") + return DatabaseOperationError("database operation failed") + + +@asynccontextmanager +async def transaction(pool: Pool) -> AsyncIterator[Connection]: + async with pool.acquire() as connection: + try: + async with connection.transaction(): + yield connection + except asyncpg.PostgresError as error: + raise map_database_error(error) from error + + +@asynccontextmanager +async def savepoint(connection: Connection) -> AsyncIterator[Connection]: + try: + async with connection.transaction(): + yield connection + except asyncpg.PostgresError as error: + raise map_database_error(error) from error + + +def require_affected(status: str, expected: int = 1) -> None: + try: + affected = int(status.rsplit(" ", 1)[1]) + except (IndexError, ValueError) as error: + raise DatabaseOperationError(f"unrecognized command status: {status}") from error + if affected != expected: + raise DatabaseOperationError(f"expected {expected} affected row(s), got {affected}") + + +@dataclass(frozen=True, slots=True) +class RetryPolicy: + attempts: int = 3 + base_delay_seconds: float = 0.02 + + +DEFAULT_RETRY_POLICY = RetryPolicy() + + +async def retry_transient[T]( + operation: Callable[[], Awaitable[T]], policy: RetryPolicy = DEFAULT_RETRY_POLICY +) -> T: + if policy.attempts < 1: + raise ValueError("retry attempts must be positive") + for attempt in range(policy.attempts): + try: + return await operation() + except TransientDatabaseError: + if attempt + 1 == policy.attempts: + raise + await asyncio.sleep(policy.base_delay_seconds * (2**attempt)) + raise RuntimeError("unreachable retry state") diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..6a33c79 --- /dev/null +++ b/tests/integration/__init__.py @@ -0,0 +1 @@ +"""PostgreSQL-backed integration tests.""" diff --git a/tests/integration/test_migrations.py b/tests/integration/test_migrations.py new file mode 100644 index 0000000..02aaffd --- /dev/null +++ b/tests/integration/test_migrations.py @@ -0,0 +1,36 @@ +"""PostgreSQL migration acceptance checks.""" + +import os +import uuid + +import asyncpg # type: ignore[import-untyped] +import pytest + +from app.db.migrations import migrate + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif(not os.getenv("TEST_DATABASE_URL"), reason="TEST_DATABASE_URL is required"), +] + + +async def test_migration_is_repeatable_and_constraints_hold() -> None: + connection = await asyncpg.connect(os.environ["TEST_DATABASE_URL"]) + try: + await migrate(connection) + assert await migrate(connection) == 0 + node_id = uuid.uuid4() + await connection.execute( + "INSERT INTO nodes(id, name, status) VALUES($1, $2, 'online') ON CONFLICT DO NOTHING", + node_id, + f"test-{node_id}", + ) + with pytest.raises(asyncpg.CheckViolationError): + async with connection.transaction(): + await connection.execute( + "INSERT INTO nodes(id, name, status) VALUES($1, $2, 'invalid')", + uuid.uuid4(), + f"invalid-{node_id}", + ) + finally: + await connection.close() diff --git a/tests/unit/test_db_primitives.py b/tests/unit/test_db_primitives.py new file mode 100644 index 0000000..9b5d78d --- /dev/null +++ b/tests/unit/test_db_primitives.py @@ -0,0 +1,59 @@ +"""Database primitive behavior independent of PostgreSQL.""" + +import asyncpg # type: ignore[import-untyped] +import pytest + +from app.db.primitives import ( + ConflictError, + DatabaseOperationError, + ReferenceError, + RetryPolicy, + TransientDatabaseError, + map_database_error, + require_affected, + retry_transient, +) + + +def test_error_mapping_is_stable_and_safe() -> None: + assert isinstance(map_database_error(asyncpg.UniqueViolationError("secret")), ConflictError) + assert isinstance( + map_database_error(asyncpg.ForeignKeyViolationError("secret")), ReferenceError + ) + assert isinstance( + map_database_error(asyncpg.SerializationError("secret")), TransientDatabaseError + ) + assert "secret" not in str(map_database_error(asyncpg.PostgresError("secret"))) + + +def test_affected_row_checks() -> None: + require_affected("UPDATE 1") + with pytest.raises(DatabaseOperationError, match="expected 1"): + require_affected("UPDATE 0") + with pytest.raises(DatabaseOperationError, match="unrecognized"): + require_affected("BROKEN") + + +async def test_transient_retry_is_bounded() -> None: + calls = 0 + + async def operation() -> str: + nonlocal calls + calls += 1 + if calls < 3: + raise TransientDatabaseError("retry") + return "ok" + + assert await retry_transient(operation, RetryPolicy(attempts=3, base_delay_seconds=0)) == "ok" + assert calls == 3 + + +async def test_transient_retry_propagates_final_failure() -> None: + async def operation() -> None: + raise TransientDatabaseError("retry") + + with pytest.raises(TransientDatabaseError): + await retry_transient(operation, RetryPolicy(attempts=2, base_delay_seconds=0)) + + with pytest.raises(ValueError, match="positive"): + await retry_transient(operation, RetryPolicy(attempts=0)) diff --git a/tests/unit/test_migrations.py b/tests/unit/test_migrations.py new file mode 100644 index 0000000..b2365a4 --- /dev/null +++ b/tests/unit/test_migrations.py @@ -0,0 +1,32 @@ +"""Migration discovery and checksum tests.""" + +from pathlib import Path + +from app.db.migrations import load_migrations + + +def test_load_migrations_is_ordered_and_checksummed(tmp_path: Path) -> None: + (tmp_path / "002_second.sql").write_text("SELECT 2;") + (tmp_path / "001_first.sql").write_text("SELECT 1;") + + migrations = load_migrations(tmp_path) + + assert [migration.version for migration in migrations] == [1, 2] + assert migrations[0].name == "001_first" + assert len(migrations[0].checksum) == 64 + + +def test_repository_migration_defines_required_planes() -> None: + migration = load_migrations()[0] + + for table in ( + "contract_snapshots", + "nodes", + "resources", + "principals", + "acl_entries", + "tasks", + "scenarios", + "audit_events", + ): + assert f"CREATE TABLE {table}" in migration.sql