feat: add migration framework with FastMigrate integration

- Add migrations.py with create_migration, run_migrations, get_db_version
- Implement CLI migrate and create-migration commands
- Use FastMigrate's sequential 0001-*.sql naming convention
- Add comprehensive unit, integration, and E2E tests (35 tests)
- Add fresh_db_path and temp_migrations_dir test fixtures

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2025-12-27 18:25:28 +00:00
parent 32bb3c01e0
commit 7d7cd2bb57
6 changed files with 756 additions and 11 deletions

View File

@@ -36,13 +36,32 @@ def main():
sys.exit(1)
if args.command == "migrate":
print(f"Running migrations (to={args.to})...")
# TODO: Implement migration runner
print("Migration not yet implemented")
from animaltrack.config import Settings
from animaltrack.migrations import run_migrations
settings = Settings()
migrations_dir = "migrations"
print(f"Running migrations from {migrations_dir}...")
success = run_migrations(
db_path=settings.db_path,
migrations_dir=migrations_dir,
verbose=True,
)
if success:
print("Migrations completed successfully")
sys.exit(0)
else:
print("Migration failed", file=sys.stderr)
sys.exit(1)
elif args.command == "create-migration":
print(f"Creating migration: {args.description}")
# TODO: Implement migration creation
print("Migration creation not yet implemented")
from animaltrack.migrations import create_migration
migrations_dir = "migrations"
filepath = create_migration(migrations_dir, args.description)
print(f"Created migration: {filepath}")
print(" Edit the file to add your schema changes")
elif args.command == "seed":
print("Loading seed data...")
# TODO: Implement seeder

View File

@@ -0,0 +1,152 @@
# ABOUTME: Migration framework wrapper for FastMigrate.
# ABOUTME: Provides migration creation, discovery, and execution with version control.
import re
from pathlib import Path
import fastmigrate
def get_next_migration_index(migrations_dir: str | Path) -> int:
"""
Discover the next migration index by scanning existing migrations.
Looks for files matching pattern: 0001-*.sql|py|sh
Returns highest index + 1, or 1 if directory is empty.
Args:
migrations_dir: Path to migrations directory
Returns:
Next sequential migration index (1-based)
"""
migrations_path = Path(migrations_dir)
if not migrations_path.exists():
return 1
# Pattern: 0001-description.ext
pattern = re.compile(r"^(\d{4})-.*\.(sql|py|sh)$")
max_index = 0
for file in migrations_path.iterdir():
if file.is_file():
if match := pattern.match(file.name):
index = int(match.group(1))
max_index = max(max_index, index)
return max_index + 1
def sanitize_description(description: str) -> str:
"""
Convert migration description to valid filename component.
- Lowercase
- Replace spaces with hyphens
- Remove non-alphanumeric except hyphens
- Collapse multiple hyphens
Args:
description: Human-readable migration description
Returns:
Sanitized filename-safe string
Examples:
"Add Users Table!" -> "add-users-table"
"Fix Bug #123" -> "fix-bug-123"
"""
# Lowercase and replace spaces with hyphens
sanitized = description.lower().replace(" ", "-")
# Keep only alphanumeric and hyphens
sanitized = re.sub(r"[^a-z0-9-]", "", sanitized)
# Collapse multiple hyphens
sanitized = re.sub(r"-+", "-", sanitized)
# Strip leading/trailing hyphens
return sanitized.strip("-")
def create_migration(migrations_dir: str | Path, description: str) -> Path:
"""
Create a new migration file with next sequential index.
Generates file: migrations/XXXX-description.sql where XXXX is 4-digit index.
Includes ABOUTME header comments.
Args:
migrations_dir: Path to migrations directory
description: Human-readable migration description
Returns:
Path to created migration file
Example:
create_migration("migrations", "add users table")
-> Path("migrations/0001-add-users-table.sql")
"""
migrations_path = Path(migrations_dir)
migrations_path.mkdir(parents=True, exist_ok=True)
index = get_next_migration_index(migrations_path)
sanitized = sanitize_description(description)
filename = f"{index:04d}-{sanitized}.sql"
filepath = migrations_path / filename
# SQL template with ABOUTME header
template = f"""-- ABOUTME: Migration {index:04d} - {description}
-- ABOUTME: Created for AnimalTrack database schema versioning.
-- Write your migration SQL here
"""
filepath.write_text(template)
return filepath
def get_db_version(db_path: str | Path) -> int:
"""
Get current database migration version.
Wraps fastmigrate.get_db_version with path conversion.
Args:
db_path: Path to SQLite database
Returns:
Current migration version (0 for new versioned DB)
Raises:
FileNotFoundError: If database doesn't exist
sqlite3.Error: If database is not versioned
"""
return fastmigrate.get_db_version(Path(db_path))
def run_migrations(
db_path: str | Path,
migrations_dir: str | Path,
verbose: bool = False,
) -> bool:
"""
Run pending database migrations.
Initializes database if not versioned. Runs migrations in sequential order.
Args:
db_path: Path to SQLite database
migrations_dir: Path to migrations directory
verbose: Print detailed progress messages
Returns:
True if all migrations succeeded, False otherwise
"""
db_path = Path(db_path)
migrations_path = Path(migrations_dir)
# Ensure database is versioned
fastmigrate.create_db(db_path)
# Run all pending migrations
return fastmigrate.run_migrations(db_path, migrations_path, verbose=verbose)