Imported from Ayub-Khan/immortal_agent_swarm (.agents/skills/migration-planning/SKILL.md). Install upstream with npx skills add Ayub-Khan/immortal_agent_swarm --skill migration-planning. Copyright stays with the author.

Migration Planning Skill

Table of Contents

1. Schema Versioning Strategies
2. Zero-Downtime Migrations
3. Rollback Procedures
4. Testing Migrations
5. Security Checklists

Schema Versioning Strategies

Timestamp-Based Versioning

-- Format: YYYYMMDDHHMMSS_description.sql
-- 20240115143000_add_user_status.sql

ALTER TABLE users ADD COLUMN status VARCHAR(20) DEFAULT 'active';

Sequential Versioning

-- Format: V{version}__{description}.sql
-- V001__create_users_table.sql
-- V002__add_user_status.sql

-- Flyway/Liquibase style
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    email VARCHAR(255) UNIQUE NOT NULL
);

Semantic Versioning

# Django migration naming
# 0001_initial.py
# 0002_add_user_status.py
# 0003_alter_user_email.py

# Version tracking in migration file
class Migration(migrations.Migration):
    version = "1.2.0"  # Semantic version
    dependencies = [
        ('app', '0002_add_user_status'),
    ]

Git-Based Versioning

# Use commit hash for tracking
MIGRATION_ID=$(git rev-parse --short HEAD)
echo "-- Migration: $MIGRATION_ID" > "migrations/${MIGRATION_ID}_schema_change.sql"

# Tag releases
git tag -a db-v1.2.0 -m "Database schema v1.2.0"

Zero-Downtime Migrations

Step 1: Assessment

def assess_migration_impact(table_name, operation):
    """Assess if migration can be zero-downtime."""
    
    safe_operations = [
        'ADD COLUMN NULLABLE',
        'CREATE INDEX CONCURRENTLY',
        'ADD CONSTRAINT NOT VALID',
    ]
    
    unsafe_operations = [
        'DROP COLUMN',
        'ALTER COLUMN TYPE',
        'ADD COLUMN DEFAULT',
        'VACUUM FULL',
    ]
    
    if operation in safe_operations:
        return {"safe": True, "strategy": "direct"}
    elif operation in unsafe_operations:
        return {"safe": False, "strategy": "expand-contract"}
    else:
        return {"safe": None, "strategy": "analyze"}

Step 2: Expand Phase (Add New)

-- PostgreSQL: Add nullable column (safe)
ALTER TABLE users ADD COLUMN email_v2 VARCHAR(255) NULL;

-- Add index concurrently (no table lock)
CREATE INDEX CONCURRENTLY idx_users_email_v2 ON users(email_v2);

-- Add constraint as NOT VALID (fast, validates later)
ALTER TABLE users ADD CONSTRAINT chk_email_format 
    CHECK (email_v2 ~* '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$') 
    NOT VALID;

Step 3: Backfill Phase (Migrate Data)

# Batch backfill to avoid long transactions
def backfill_batch(batch_size=1000):
    """Backfill data in small batches."""
    while True:
        with connection.cursor() as cursor:
            cursor.execute("""
                UPDATE users 
                SET email_v2 = email
                WHERE id IN (
                    SELECT id FROM users 
                    WHERE email_v2 IS NULL 
                    LIMIT %s
                    FOR UPDATE SKIP LOCKED
                )
            """, [batch_size])
            
            if cursor.rowcount == 0:
                break
            
            # Brief pause to let other transactions through
            time.sleep(0.1)

Step 4: Validate Phase (Ensure Integrity)

-- Validate constraint (can be done online in PostgreSQL)
ALTER TABLE users VALIDATE CONSTRAINT chk_email_format;

-- Verify backfill completeness
SELECT COUNT(*) FROM users WHERE email_v2 IS NULL;
-- Should return 0

Step 5: Contract Phase (Remove Old)

# Application code transition
class User(models.Model):
    email = models.EmailField()      # Old field
    email_v2 = models.EmailField(null=True)  # New field
    
    def save(self, *args, **kwargs):
        # Dual write during transition
        self.email_v2 = self.email
        super().save(*args, **kwargs)
    
    @property
    def get_email(self):
        # Read from new field after transition
        return self.email_v2 or self.email

-- Final cleanup (after application deployed)
ALTER TABLE users DROP COLUMN email;
ALTER TABLE users RENAME COLUMN email_v2 TO email;
ALTER TABLE users ALTER COLUMN email SET NOT NULL;

Framework-Specific Patterns

Django

# Migration 1: Add nullable field
class Migration(migrations.Migration):
    operations = [
        migrations.AddField(
            model_name='user',
            name='email_v2',
            field=models.EmailField(null=True),
        ),
    ]

# Migration 2: Backfill (separate transaction)
class Migration(migrations.Migration):
    atomic = False  # Allow other queries during migration
    
    operations = [
        migrations.RunPython(backfill_email_v2, reverse_code=migrations.RunPython.noop),
    ]

# Migration 3: Make required and drop old
class Migration(migrations.Migration):
    operations = [
        migrations.RemoveField(model_name='user', name='email'),
        migrations.RenameField(model_name='user', old_name='email_v2', new_name='email'),
        migrations.AlterField(
            model_name='user',
            name='email',
            field=models.EmailField(null=False),
        ),
    ]

SQLAlchemy/Alembic

# alembic migration
from alembic import op
import sqlalchemy as sa

# Revision identifiers
revision = 'abc123'
down_revision = 'def456'

def upgrade():
    # Step 1: Add new column
    op.add_column('users', sa.Column('email_v2', sa.String(255), nullable=True))
    
    # Step 2: Create index concurrently
    op.execute('CREATE INDEX CONCURRENTLY idx_email_v2 ON users(email_v2)')
    
    # Step 3: Backfill (batch operation)
    op.execute("""
        UPDATE users 
        SET email_v2 = email 
        WHERE email_v2 IS NULL
    """)
    
    # Step 4: Make required
    op.alter_column('users', 'email_v2', nullable=False)
    
    # Step 5: Drop old
    op.drop_column('users', 'email')
    op.rename_column('users', 'email_v2', 'email')

def downgrade():
    # Reverse operations
    op.add_column('users', sa.Column('email', sa.String(255), nullable=True))
    op.execute("UPDATE users SET email = email_v2")
    op.drop_column('users', 'email_v2')

Rollback Procedures

Strategy 1: Migration-Level Rollback

# Always implement reverse operations
class Migration(migrations.Migration):
    
    def forward_migration(apps, schema_editor):
        User = apps.get_model('app', 'User')
        User.objects.filter(status__isnull=True).update(status='active')
    
    def reverse_migration(apps, schema_editor):
        User = apps.get_model('app', 'User')
        User.objects.filter(status='active').update(status=None)
    
    operations = [
        migrations.RunPython(forward_migration, reverse_migration),
    ]

Strategy 2: Database Snapshot Rollback

#!/bin/bash
# rollback_snapshot.sh

TIMESTAMP=$(date +%Y%m%d_%H%M%S)
DB_NAME="production"
BACKUP_FILE="backups/pre_migration_${TIMESTAMP}.sql"

echo "Creating backup..."
pg_dump -h $DB_HOST -U $DB_USER -d $DB_NAME > $BACKUP_FILE

echo "Running migration..."
if python manage.py migrate; then
    echo "Migration successful"
else
    echo "Migration failed! Rolling back..."
    psql -h $DB_HOST -U $DB_USER -d $DB_NAME < $BACKUP_FILE
    exit 1
fi

Strategy 3: Feature Flag Rollback

# settings.py
USE_NEW_SCHEMA = env.bool('USE_NEW_SCHEMA', default=False)

# models.py
class User(models.Model):
    email = models.EmailField()
    email_v2 = models.EmailField(null=True)
    
    @property
    def effective_email(self):
        if settings.USE_NEW_SCHEMA and self.email_v2:
            return self.email_v2
        return self.email

# Rollback: Just change environment variable
# USE_NEW_SCHEMA=false

Rollback Decision Matrix

Testing Migrations

Step 1: Pre-Migration Testing

# test_migration_safety.py
import pytest
from django.db import connection
from django.core.management import call_command

def test_migration_sql_syntax():
    """Verify migration SQL is valid."""
    with connection.cursor() as cursor:
        # Get SQL without executing
        call_command('sqlmigrate', 'app_name', '0002', stdout=open('/tmp/migration.sql', 'w'))
        
        # Parse and validate
        with open('/tmp/migration.sql') as f:
            sql = f.read()
            
        # Check for dangerous operations
        dangerous = ['DROP TABLE', 'DELETE FROM', 'TRUNCATE']
        for op in dangerous:
            if op in sql.upper():
                pytest.fail(f"Dangerous operation found: {op}")

def test_migration_reversibility():
    """Ensure migration can be reversed."""
    # Apply
    call_command('migrate', 'app_name', '0002')
    
    # Reverse
    try:
        call_command('migrate', 'app_name', '0001')
    except Exception as e:
        pytest.fail(f"Migration not reversible: {e}")

Step 2: Staging Environment Testing

# staging_migration_test.py
import subprocess
import time

def test_migration_on_staging():
    """Test migration against staging database."""
    
    # 1. Restore production snapshot to staging
    subprocess.run([
        'pg_restore', 
        '-h', 'staging-db', 
        '-d', 'staging_db', 
        'production_snapshot.sql'
    ], check=True)
    
    # 2. Record pre-migration state
    pre_counts = get_table_counts('staging_db')
    
    # 3. Apply migration
    start_time = time.time()
    result = subprocess.run(
        ['python', 'manage.py', 'migrate'],
        capture_output=True,
        text=True
    )
    duration = time.time() - start_time
    
    # 4. Verify
    assert result.returncode == 0, f"Migration failed: {result.stderr}"
    assert duration < 300, f"Migration too slow: {duration}s"
    
    # 5. Data integrity check
    post_counts = get_table_counts('staging_db')
    assert pre_counts == post_counts, "Row counts changed!"
    
    return {"success": True, "duration": duration}

Step 3: Data Integrity Testing

# migration_integrity_checks.py

def run_integrity_checks():
    """Run post-migration data integrity checks."""
    checks = []
    
    with connection.cursor() as cursor:
        # Check 1: No orphaned foreign keys
        cursor.execute("""
            SELECT COUNT(*) FROM orders o
            LEFT JOIN users u ON o.user_id = u.id
            WHERE u.id IS NULL AND o.user_id IS NOT NULL
        """)
        orphaned = cursor.fetchone()[0]
        checks.append({"name": "orphaned_orders", "passed": orphaned == 0, "count": orphaned})
        
        # Check 2: Required fields populated
        cursor.execute("""
            SELECT COUNT(*) FROM users WHERE email IS NULL
        """)
        null_emails = cursor.fetchone()[0]
        checks.append({"name": "null_emails", "passed": null_emails == 0, "count": null_emails})
        
        # Check 3: Unique constraints
        cursor.execute("""
            SELECT email, COUNT(*) FROM users 
            GROUP BY email 
            HAVING COUNT(*) > 1
        """)
        duplicates = cursor.fetchall()
        checks.append({"name": "duplicate_emails", "passed": len(duplicates) == 0, "duplicates": duplicates})
        
        # Check 4: Row count sanity
        cursor.execute("SELECT COUNT(*) FROM users")
        user_count = cursor.fetchone()[0]
        checks.append({"name": "user_count", "passed": user_count > 0, "count": user_count})
    
    return checks

Step 4: Performance Testing

# migration_performance_test.py
import time
from django.db import connection

def test_migration_performance():
    """Ensure migration doesn't cause performance degradation."""
    
    # Baseline queries
    baseline_queries = [
        ("SELECT COUNT(*) FROM users", "user_count"),
        ("SELECT * FROM users WHERE email = 'test@example.com'", "user_lookup"),
        ("SELECT * FROM orders JOIN users ON orders.user_id = users.id LIMIT 100", "join_query"),
    ]
    
    results = []
    for query, name in baseline_queries:
        times = []
        for _ in range(5):
            start = time.time()
            with connection.cursor() as cursor:
                cursor.execute(query)
            times.append(time.time() - start)
        
        avg_time = sum(times) / len(times)
        results.append({
            "query": name,
            "avg_time_ms": avg_time * 1000,
            "acceptable": avg_time < 0.1  # 100ms threshold
        })
    
    return results

Security Checklists

Pre-Migration Security Checklist

• Backup Verified: Production backup taken and restore tested
• Access Controls: Migration run with least-privilege account
• Data Classification: Sensitive data handling documented
• Encryption: Data encrypted in transit and at rest
• Audit Logging: All changes will be logged
• PII Handling: No PII exposed in migration logs
• Compliance: Changes reviewed for SOC2/ISO27001 compliance
• Approval: Change approved by security team

Migration Security Checklist

• Connection Security: SSL/TLS enabled for database connections
• Credential Management: No hardcoded credentials in migration files
• SQL Injection: All dynamic SQL properly parameterized
• Privilege Escalation: No GRANT/REVOKE operations without approval
• Data Masking: Sensitive data masked in non-production environments

Post-Migration Security Checklist

• Access Review: Verify user permissions unchanged
• Audit Log Review: Check for anomalies in migration logs
• Vulnerability Scan: Run security scanner on updated schema
• Data Integrity: Verify no unauthorized data modifications
• Secret Rotation: Rotate any credentials used during migration
• Documentation: Security implications documented

Data Protection Checklist

• Encryption at Rest: Verify tablespace encryption still active
• Column Encryption: Encrypted columns still functional
• Data Masking: Masking rules preserved
• Retention Policies: Data retention rules unchanged
• Right to Deletion: Deletion mechanisms still work

Quick Reference

Migration Sizing Guide

PostgreSQL Lock-Safe Operations

-- Safe (minimal locking)
CREATE INDEX CONCURRENTLY ...
ALTER TABLE ... ADD COLUMN ... NULL
ALTER TABLE ... ADD CONSTRAINT ... NOT VALID
ALTER TABLE ... VALIDATE CONSTRAINT ...

-- Unsafe (exclusive lock)
ALTER TABLE ... DROP COLUMN
ALTER TABLE ... ALTER COLUMN TYPE
ALTER TABLE ... ADD COLUMN ... DEFAULT ... NOT NULL
VACUUM FULL

Rollback Commands by Framework

# Django
python manage.py migrate app_name 0001  # Rollback to specific
python manage.py migrate app_name zero  # Rollback all

# Alembic
alembic downgrade -1           # Rollback one
alembic downgrade base         # Rollback all
alembic downgrade abc123       # Rollback to revision

# Flyway
flyway undo                    # Undo last migration
flyway repair                  # Fix migration state

# Prisma
npx prisma migrate resolve --rolled-back "migration_name"

Last updated: 2026-04-08