refactoring-backend #3
@@ -364,6 +364,91 @@ assert escape_like("10.0.0.1") == "10.0.0.1" # Unchanged
|
||||
|
||||
---
|
||||
|
||||
## 6.2 Database Migrations
|
||||
|
||||
The application database schema is versioned and migrated automatically on startup via `app.db.init_db()`.
|
||||
|
||||
### Migration Design Principles
|
||||
|
||||
**Migrations must be atomic.** All schema changes for a single version (DDL statements) and the `schema_migrations` record insert must be wrapped in a single `BEGIN IMMEDIATE ... COMMIT` transaction. This prevents partial migrations if a process crashes mid-migration.
|
||||
|
||||
If a crash occurs between migration steps, the next startup will:
|
||||
1. Detect the missing `schema_migrations` record.
|
||||
2. Re-apply the entire migration in a single transaction (all-or-nothing).
|
||||
3. Avoid data corruption or schema inconsistency.
|
||||
|
||||
### Writing a New Migration
|
||||
|
||||
1. **Add the DDL statements** to `_MIGRATIONS` dict in `app/db.py`:
|
||||
|
||||
```python
|
||||
_MIGRATIONS: dict[int, str] = {
|
||||
1: _CREATE_INITIAL_SCHEMA,
|
||||
2: """
|
||||
-- Migration 2: Add new_column to users table.
|
||||
ALTER TABLE users ADD COLUMN new_column TEXT DEFAULT 'default_value';
|
||||
CREATE INDEX idx_users_new_column ON users(new_column);
|
||||
""",
|
||||
}
|
||||
```
|
||||
|
||||
2. **Update `_CURRENT_SCHEMA_VERSION`** to the new version number:
|
||||
|
||||
```python
|
||||
_CURRENT_SCHEMA_VERSION: int = 2 # was 1
|
||||
```
|
||||
|
||||
3. **Ensure idempotency where possible:**
|
||||
- Use `CREATE TABLE IF NOT EXISTS` and `CREATE INDEX IF NOT EXISTS`.
|
||||
- For `ALTER TABLE ADD COLUMN`, check if the column exists first using `PRAGMA table_info()` if re-applying the migration is a concern.
|
||||
|
||||
4. **Verify atomicity in tests:**
|
||||
|
||||
```python
|
||||
async def test_migration_2_is_atomic(tmp_path: Path) -> None:
|
||||
"""Verify migration 2 rolls back on failure."""
|
||||
db = await open_db(str(tmp_path / "test.db"))
|
||||
try:
|
||||
await db.execute("CREATE TABLE schema_migrations (version INTEGER PRIMARY KEY);")
|
||||
await db.commit()
|
||||
|
||||
# Add a test migration that fails mid-way
|
||||
original = db_module._MIGRATIONS.copy()
|
||||
db_module._MIGRATIONS[99] = """
|
||||
CREATE TABLE test_table (id INTEGER PRIMARY KEY);
|
||||
INSERT INTO nonexistent_table VALUES (1);
|
||||
"""
|
||||
|
||||
try:
|
||||
with pytest.raises(Exception):
|
||||
await _apply_migration(db, 99)
|
||||
|
||||
# Verify rollback: migration NOT recorded
|
||||
async with db.execute(
|
||||
"SELECT version FROM schema_migrations WHERE version = 99;"
|
||||
) as cursor:
|
||||
assert await cursor.fetchone() is None
|
||||
|
||||
# Verify rollback: table NOT created
|
||||
async with db.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name='test_table';"
|
||||
) as cursor:
|
||||
assert await cursor.fetchone() is None
|
||||
finally:
|
||||
db_module._MIGRATIONS = original
|
||||
finally:
|
||||
await db.close()
|
||||
```
|
||||
|
||||
### Common Pitfalls
|
||||
|
||||
- **Non-idempotent statements** — `ALTER TABLE ADD COLUMN` without `IF NOT EXISTS` will fail on re-run. Use explicit checks if needed.
|
||||
- **Comments containing semicolons** — the migration parser strips comments correctly, but avoid unusual comment syntax.
|
||||
- **String literals with semicolons** — the parser handles these; no special escaping needed.
|
||||
- **Multiple operations in one migration** — keep migrations focused. Combine related DDL but split unrelated changes.
|
||||
|
||||
---
|
||||
|
||||
## 7. Logging
|
||||
|
||||
- Use **structlog** for every log message.
|
||||
|
||||
@@ -142,7 +142,7 @@ async def _configure_connection(db: aiosqlite.Connection) -> None:
|
||||
|
||||
async def _get_current_schema_version(db: aiosqlite.Connection) -> int:
|
||||
"""Return the highest applied schema version for the given database."""
|
||||
await db.executescript(_CREATE_SCHEMA_MIGRATIONS)
|
||||
await db.execute(_CREATE_SCHEMA_MIGRATIONS)
|
||||
async with db.execute("SELECT MAX(version) FROM schema_migrations;") as cursor:
|
||||
row = await cursor.fetchone()
|
||||
if row is None or row[0] is None:
|
||||
@@ -150,12 +150,114 @@ async def _get_current_schema_version(db: aiosqlite.Connection) -> int:
|
||||
return int(row[0])
|
||||
|
||||
|
||||
async def _parse_migration_statements(script: str) -> list[str]:
|
||||
"""Parse a migration script into individual SQL statements.
|
||||
|
||||
Splits on semicolons but ignores semicolons inside string literals and
|
||||
comments. Handles both block (-- comment) and line comments.
|
||||
|
||||
Args:
|
||||
script: The raw migration script.
|
||||
|
||||
Returns:
|
||||
List of SQL statements (stripped of whitespace and comments).
|
||||
"""
|
||||
statements: list[str] = []
|
||||
current_stmt: list[str] = []
|
||||
i = 0
|
||||
|
||||
while i < len(script):
|
||||
char = script[i]
|
||||
|
||||
# Skip block comments (-- ...)
|
||||
if i < len(script) - 1 and script[i:i+2] == "--":
|
||||
while i < len(script) and script[i] != "\n":
|
||||
i += 1
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# Skip line comments (/* ... */)
|
||||
if i < len(script) - 1 and script[i:i+2] == "/*":
|
||||
i += 2
|
||||
while i < len(script) - 1:
|
||||
if script[i:i+2] == "*/":
|
||||
i += 2
|
||||
break
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# Handle string literals (single or double quotes)
|
||||
if char in ("'", '"'):
|
||||
quote = char
|
||||
current_stmt.append(char)
|
||||
i += 1
|
||||
while i < len(script):
|
||||
if script[i] == quote:
|
||||
if i + 1 < len(script) and script[i + 1] == quote:
|
||||
# Escaped quote
|
||||
current_stmt.append(quote)
|
||||
current_stmt.append(quote)
|
||||
i += 2
|
||||
else:
|
||||
# End of string
|
||||
current_stmt.append(quote)
|
||||
i += 1
|
||||
break
|
||||
else:
|
||||
current_stmt.append(script[i])
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# Statement separator
|
||||
if char == ";":
|
||||
stmt = "".join(current_stmt).strip()
|
||||
if stmt:
|
||||
statements.append(stmt)
|
||||
current_stmt = []
|
||||
i += 1
|
||||
continue
|
||||
|
||||
current_stmt.append(char)
|
||||
i += 1
|
||||
|
||||
# Add any remaining statement
|
||||
stmt = "".join(current_stmt).strip()
|
||||
if stmt:
|
||||
statements.append(stmt)
|
||||
|
||||
return statements
|
||||
|
||||
|
||||
async def _apply_migration(db: aiosqlite.Connection, version: int) -> None:
|
||||
"""Apply a single migration step and record its completion."""
|
||||
"""Apply a single migration step and record its completion atomically.
|
||||
|
||||
Wraps all DDL statements and the schema_migrations insert in a single
|
||||
BEGIN IMMEDIATE ... COMMIT transaction to ensure atomicity. If any
|
||||
statement fails, the entire migration is rolled back.
|
||||
|
||||
Args:
|
||||
db: An open aiosqlite.Connection.
|
||||
version: The migration version number.
|
||||
|
||||
Raises:
|
||||
Any exception from executing the migration statements or inserting
|
||||
the schema migration record will cause a rollback.
|
||||
"""
|
||||
migration_script = _MIGRATIONS[version]
|
||||
await db.executescript(migration_script)
|
||||
await db.execute("INSERT INTO schema_migrations (version) VALUES (?);", (version,))
|
||||
await db.commit()
|
||||
statements = await _parse_migration_statements(migration_script)
|
||||
|
||||
try:
|
||||
await db.execute("BEGIN IMMEDIATE;")
|
||||
|
||||
for statement in statements:
|
||||
await db.execute(statement)
|
||||
|
||||
await db.execute("INSERT INTO schema_migrations (version) VALUES (?);", (version,))
|
||||
|
||||
await db.commit()
|
||||
except Exception:
|
||||
await db.rollback()
|
||||
raise
|
||||
|
||||
|
||||
async def _migrate_schema(db: aiosqlite.Connection) -> None:
|
||||
|
||||
@@ -1,9 +1,16 @@
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import aiosqlite
|
||||
import pytest
|
||||
|
||||
from app.db import open_db
|
||||
from app.db import (
|
||||
_apply_migration,
|
||||
_parse_migration_statements,
|
||||
init_db,
|
||||
open_db,
|
||||
)
|
||||
|
||||
|
||||
async def test_open_db_applies_hardening_pragmas(tmp_path: Path) -> None:
|
||||
@@ -56,3 +63,181 @@ async def test_open_db_respects_busy_timeout_for_concurrent_writes(tmp_path: Pat
|
||||
assert row is not None and row[0] == "locked"
|
||||
finally:
|
||||
await connection_a.close()
|
||||
|
||||
|
||||
async def test_parse_migration_statements_single_statement() -> None:
|
||||
"""Test parsing a single statement without comments."""
|
||||
script = "CREATE TABLE test (id INTEGER PRIMARY KEY);"
|
||||
statements = await _parse_migration_statements(script)
|
||||
assert len(statements) == 1
|
||||
assert statements[0] == "CREATE TABLE test (id INTEGER PRIMARY KEY)"
|
||||
|
||||
|
||||
async def test_parse_migration_statements_multiple_statements() -> None:
|
||||
"""Test parsing multiple statements separated by semicolons."""
|
||||
script = """
|
||||
CREATE TABLE test (id INTEGER PRIMARY KEY);
|
||||
CREATE INDEX idx_test ON test(id);
|
||||
"""
|
||||
statements = await _parse_migration_statements(script)
|
||||
assert len(statements) == 2
|
||||
assert "CREATE TABLE test" in statements[0]
|
||||
assert "CREATE INDEX idx_test" in statements[1]
|
||||
|
||||
|
||||
async def test_parse_migration_statements_with_line_comments() -> None:
|
||||
"""Test parsing statements with line comments."""
|
||||
script = """
|
||||
-- This is a comment
|
||||
CREATE TABLE test (id INTEGER PRIMARY KEY);
|
||||
-- Another comment
|
||||
"""
|
||||
statements = await _parse_migration_statements(script)
|
||||
assert len(statements) == 1
|
||||
assert "CREATE TABLE test" in statements[0]
|
||||
|
||||
|
||||
async def test_parse_migration_statements_with_block_comments() -> None:
|
||||
"""Test parsing statements with block comments."""
|
||||
script = """
|
||||
/* Block comment */
|
||||
CREATE TABLE test (id INTEGER PRIMARY KEY);
|
||||
/* Another block */
|
||||
"""
|
||||
statements = await _parse_migration_statements(script)
|
||||
assert len(statements) == 1
|
||||
assert "CREATE TABLE test" in statements[0]
|
||||
|
||||
|
||||
async def test_parse_migration_statements_with_string_literals() -> None:
|
||||
"""Test parsing statements with string literals containing semicolons."""
|
||||
script = """
|
||||
CREATE TABLE test (id INTEGER PRIMARY KEY, data TEXT);
|
||||
INSERT INTO test (data) VALUES ('value; with; semicolons');
|
||||
"""
|
||||
statements = await _parse_migration_statements(script)
|
||||
assert len(statements) == 2
|
||||
assert "CREATE TABLE test" in statements[0]
|
||||
assert "INSERT INTO test" in statements[1]
|
||||
assert "value; with; semicolons" in statements[1]
|
||||
|
||||
|
||||
async def test_parse_migration_statements_with_escaped_quotes() -> None:
|
||||
"""Test parsing statements with escaped quotes in string literals."""
|
||||
script = """
|
||||
INSERT INTO test (data) VALUES ('it''s a test');
|
||||
"""
|
||||
statements = await _parse_migration_statements(script)
|
||||
assert len(statements) == 1
|
||||
assert "it''s a test" in statements[0]
|
||||
|
||||
|
||||
async def test_apply_migration_is_atomic_success(tmp_path: Path) -> None:
|
||||
"""Test that migration is atomic when all statements succeed."""
|
||||
database_path = str(tmp_path / "bangui_atomic.db")
|
||||
db = await open_db(database_path)
|
||||
try:
|
||||
# Initialize schema_migrations table
|
||||
await db.execute(
|
||||
"CREATE TABLE IF NOT EXISTS schema_migrations (version INTEGER PRIMARY KEY, migrated_at TEXT);"
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
# Apply a test migration
|
||||
await _apply_migration(db, 1)
|
||||
|
||||
# Verify the migration was recorded
|
||||
async with db.execute(
|
||||
"SELECT version FROM schema_migrations WHERE version = 1;"
|
||||
) as cursor:
|
||||
row = await cursor.fetchone()
|
||||
assert row is not None and row[0] == 1
|
||||
|
||||
# Verify the schema tables exist
|
||||
async with db.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name='settings';"
|
||||
) as cursor:
|
||||
row = await cursor.fetchone()
|
||||
assert row is not None
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
async def test_apply_migration_is_atomic_rollback(tmp_path: Path) -> None:
|
||||
"""Test that migration is rolled back when a statement fails.
|
||||
|
||||
This test verifies that when an error occurs mid-migration, the
|
||||
transaction is rolled back and the schema_migrations table is NOT updated.
|
||||
"""
|
||||
database_path = str(tmp_path / "bangui_rollback.db")
|
||||
db = await open_db(database_path)
|
||||
try:
|
||||
# Initialize schema_migrations table
|
||||
await db.execute(
|
||||
"CREATE TABLE IF NOT EXISTS schema_migrations (version INTEGER PRIMARY KEY, migrated_at TEXT);"
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
# Create a custom migration that will fail
|
||||
from app import db as db_module
|
||||
|
||||
original_migrations = db_module._MIGRATIONS.copy()
|
||||
|
||||
# Add a migration that will fail on the second statement
|
||||
db_module._MIGRATIONS[99] = """
|
||||
CREATE TABLE test_rollback (id INTEGER PRIMARY KEY);
|
||||
INSERT INTO nonexistent_table VALUES (1);
|
||||
"""
|
||||
|
||||
try:
|
||||
# Attempt migration; it should fail
|
||||
with pytest.raises(Exception): # sqlite3 will raise an error
|
||||
await _apply_migration(db, 99)
|
||||
|
||||
# Verify the migration was NOT recorded
|
||||
async with db.execute(
|
||||
"SELECT version FROM schema_migrations WHERE version = 99;"
|
||||
) as cursor:
|
||||
row = await cursor.fetchone()
|
||||
assert row is None
|
||||
|
||||
# Verify the test table was NOT created (rollback occurred)
|
||||
async with db.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name='test_rollback';"
|
||||
) as cursor:
|
||||
row = await cursor.fetchone()
|
||||
assert row is None
|
||||
finally:
|
||||
# Restore original migrations
|
||||
db_module._MIGRATIONS = original_migrations
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
async def test_init_db_idempotent(tmp_path: Path) -> None:
|
||||
"""Test that init_db is idempotent."""
|
||||
database_path = str(tmp_path / "bangui_idempotent.db")
|
||||
db = await open_db(database_path)
|
||||
try:
|
||||
# Initialize once
|
||||
await init_db(db)
|
||||
|
||||
# Get schema version
|
||||
async with db.execute(
|
||||
"SELECT MAX(version) FROM schema_migrations;"
|
||||
) as cursor:
|
||||
row1 = await cursor.fetchone()
|
||||
|
||||
# Initialize again (should be no-op)
|
||||
await init_db(db)
|
||||
|
||||
# Verify schema version is unchanged
|
||||
async with db.execute(
|
||||
"SELECT MAX(version) FROM schema_migrations;"
|
||||
) as cursor:
|
||||
row2 = await cursor.fetchone()
|
||||
|
||||
assert row1 == row2
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user