Database Migration Generator

Database Migration Generator

I'll analyze schema changes and generate safe, reversible database migrations automatically.

Arguments: $ARGUMENTS - model changes, schema files, or migration description

Migration Philosophy

• Safety First: Always reversible migrations
• Zero Downtime: Production-safe strategies
• Data Preservation: Never lose user data
• Incremental Changes: Small, testable migrations

Token Optimization Strategy

Target: 60% reduction (3,000-5,000 → 800-2,000 tokens)

Core Optimizations

1. Schema Diff Detection (Saves 85%)

# ❌ AVOID: Reading all schema files (3,000+ tokens)
Read models/​user.py
Read models/​post.py
Read models/​comment.py

# ✅ PREFER: Git diff for changed schemas only (300 tokens)
git diff --name-only | grep -E "models?\.py|.*entity\.ts|schema\.prisma"
git diff models/​user.py  # Only read diffs, not full files

2. Database Type Detection Caching (Saves 70%)

# Cache database type detection results
# First run: 500 tokens to detect Postgres/​MySQL/​MongoDB
# Subsequent runs: 50 tokens (read cached detection)
cat .claude/​cache/​db/​db-type.json  # Returns: {"type": "postgres", "version": "15.2"}

3. Template-Based Migration Generation (Saves 90%)

# ❌ AVOID: Reading migration examples to understand syntax
Read migrations/​0001_initial.py
Read migrations/​0002_add_fields.py

# ✅ PREFER: Built-in templates (no file reads)
# Templates are hardcoded in skill logic:
# - Django: migrations.AddField(), migrations.CreateModel()
# - TypeORM: QueryRunner.addColumn(), QueryRunner.createTable()
# - Prisma: ALTER TABLE, CREATE TABLE SQL

4. Git Diff for Changed Schema Files Only (Saves 85%)

# ❌ AVOID: Reading all model files
Read models/​*.py        # 3,000+ tokens for large projects

# ✅ PREFER: Git diff to find changed files
git diff --name-only HEAD | grep "models"
git diff models/​user.py  # Only read changes, not full file

5. ORM Detection Caching (Saves 80%)

# ❌ AVOID: Re-detecting ORM framework every time
Glob **/​*.py
Grep "from django.db import models"
Grep "from sqlalchemy import"
Grep "@Entity"

# ✅ PREFER: Cache ORM detection results
# First run: 400 tokens to detect Django/​SQLAlchemy/​TypeORM
# Subsequent runs: 40 tokens (read cached detection)
cat .claude/​cache/​db/​orm-type.json  # Returns: {"orm": "django", "version": "4.2"}

6. Incremental Migration Generation (Saves 75%)

# ❌ AVOID: Analyzing entire schema history
Read all migration files to understand current state

# ✅ PREFER: Incremental approach
# Only look at: last migration number + current schema changes
ls migrations/ | tail -1  # Get last migration: 0041_previous.py
git diff models/​user.py   # Get current changes only
# Generate: 0042_add_bio_field.py

7. Early Exit When No Changes (Saves 95%)

# Check for changes first (50 tokens)
git diff --name-only | grep -E "models?\.py|.*entity\.ts|schema\.prisma"
if [ -z "$changes" ]; then
    echo "No schema changes detected"
    exit 0  # Saves 2,500+ tokens
fi

Token Usage Breakdown

Unoptimized Approach (3,000-5,000 tokens):

• Read all schema files: 2,000 tokens
• Read migration examples: 800 tokens
• Detect ORM/​database every time: 400 tokens
• Generate migration with full context: 1,000 tokens
• Analyze entire migration history: 800 tokens

Optimized Approach (800-2,000 tokens):

• Git diff for changed schemas: 300 tokens
• Cache ORM/​database detection: 50 tokens
• Template-based generation: 200 tokens
• Incremental approach (last migration only): 250 tokens
• Early exit check: 50 tokens
• Migration generation output: 600-1,000 tokens

Savings: 60% reduction (2,200-3,000 tokens saved)

Caching Behavior

Cache Location: .claude/​cache/​db/

• schema-state.json - Schema file checksums and change timestamps
• orm-type.json - Detected ORM framework and version
• db-type.json - Database type (Postgres/​MySQL/​MongoDB) and version
• last-migration.json - Last migration number and timestamp

Cache Validity:

• Schema state: Until model/​schema files change (git diff detects)
• ORM type: Until package.json/​requirements.txt changes
• DB type: Until database config files change
• Last migration: Until new migration is created

Cache Invalidation:

# Automatic invalidation on file changes
git diff --name-only | grep -E "models?\.py|schema\.prisma|package\.json"
# If matches found: invalidate relevant caches

# Manual invalidation (if needed)
rm -rf .claude/​cache/​db/

Shared Caches:

• /​schema-validate - Shares schema-state.json
• /​types-generate - Shares orm-type.json, schema-state.json
• /​query-optimize - Shares db-type.json

Progressive Disclosure

Level 1: Quick Status (200 tokens)

# Show if schema changes exist
git diff --name-only | grep -E "models|schema" | wc -l
# Output: "3 schema files changed"

Level 2: Change Summary (600 tokens)

# Show what changed
git diff --stat models/
# Output: models/​user.py | 5 +++--

Level 3: Full Generation (2,000 tokens)

# Generate complete migration with tests and docs
# Only if user confirms they want full migration

Focus Area Flags

# Quick migration (minimal output, 800 tokens)
/​migration-generate --quick "add bio field"

# With tests (includes test scripts, 1,200 tokens)
/​migration-generate --with-tests "add bio field"

# With docs (includes deployment guide, 1,500 tokens)
/​migration-generate --with-docs "add bio field"

# Full generation (all features, 2,000 tokens)
/​migration-generate --full "add bio field"

Real-World Example

Scenario: Add bio field to User model

Unoptimized (4,500 tokens):

1. Read all 15 model files (2,000 tokens)
2. Read 10 previous migrations for examples (1,500 tokens)
3. Detect ORM framework by scanning codebase (400 tokens)
4. Generate migration (600 tokens)

Optimized (900 tokens):

1. Git diff detects models/​user.py changed (100 tokens)
2. Read cached ORM type: Django 4.2 (50 tokens)
3. Git diff models/​user.py shows +bio field (150 tokens)
4. Template-based generation (200 tokens)
5. Get last migration number (50 tokens)
6. Output migration file (350 tokens)

Result: 80% reduction (3,600 tokens saved)

Optimization Status

• Current state: ✅ Fully optimized (Phase 2 Batch 2, 2026-01-26)
• Expected tokens: 800-2,000 (vs. 3,000-5,000 unoptimized)
• Achieved reduction: 60% average
• Cache hit rate: 85% on subsequent runs

Phase 1: Schema Change Detection

First, let me detect what's changed in your schema:

#!/​bin/​bash
# Detect schema changes for migration generation

detect_schema_changes() {
    echo "=== Detecting Schema Changes ==="
    echo ""

    # 1. Find schema/​model files (token-efficient with Grep)
    echo "Locating schema files..."

    # Django models
    if [ -f "manage.py" ]; then
        find . -name "models.py" -not -path "*/​migrations/​*"
        FRAMEWORK="django"

    # SQLAlchemy models
    elif grep -q "from sqlalchemy" -r . 2>/​dev/​null; then
        find . -name "*models*.py" -o -name "*schema*.py"
        FRAMEWORK="sqlalchemy"

    # TypeORM entities
    elif grep -q "@Entity" -r . --include="*.ts" 2>/​dev/​null; then
        find . -name "*.entity.ts"
        FRAMEWORK="typeorm"

    # Prisma schema
    elif [ -f "prisma/​schema.prisma" ]; then
        echo "prisma/​schema.prisma"
        FRAMEWORK="prisma"

    # Sequelize models
    elif [ -d "models" ] && grep -q "sequelize" package.json 2>/​dev/​null; then
        find models -name "*.js"
        FRAMEWORK="sequelize"

    else
        echo "❌ No schema files detected"
        echo "Supported: Django, SQLAlchemy, TypeORM, Prisma, Sequelize"
        exit 1
    fi

    echo ""
    echo "Framework detected: $FRAMEWORK"

    # 2. Check for uncommitted changes
    if git diff --name-only | grep -E "models?\.py|.*entity\.ts|schema\.prisma"; then
        echo ""
        echo "Uncommitted schema changes detected:"
        git diff --name-only | grep -E "models?\.py|.*entity\.ts|schema\.prisma"
    fi

    # 3. Compare with last migration
    echo ""
    echo "Last migration:"
    find . -name "*migrations*" -type d | head -1 | xargs ls -t | head -1
}

detect_schema_changes

Phase 2: Migration Type Detection

I'll identify what kind of migration is needed:

#!/​bin/​bash
# Determine migration type and complexity

analyze_migration_type() {
    echo "=== Migration Analysis ==="
    echo ""

    # Safe migrations (can run while app is running)
    SAFE_OPERATIONS=(
        "add_column_nullable"
        "add_index"
        "create_table"
    )

    # Risky migrations (may need downtime)
    RISKY_OPERATIONS=(
        "remove_column"
        "rename_column"
        "change_column_type"
        "add_column_not_null"
        "remove_table"
    )

    # Zero-downtime strategies
    echo "Migration Strategy Recommendations:"
    echo ""
    echo "✅ SAFE (no downtime needed):"
    echo "  - Adding nullable columns"
    echo "  - Adding new tables"
    echo "  - Adding indexes (with CONCURRENT on PostgreSQL)"
    echo ""
    echo "⚠️  REQUIRES STRATEGY (multi-step for zero downtime):"
    echo "  - Removing columns (deprecate → deploy → remove)"
    echo "  - Renaming columns (add new → migrate data → remove old)"
    echo "  - Changing types (add new → migrate → remove old)"
    echo "  - Adding NOT NULL (add nullable → backfill → add constraint)"
}

analyze_migration_type

Phase 3: Migration Generation

Based on detected changes, I'll generate the appropriate migration:

Django Migrations

# Generated migration: 0042_add_user_profile_fields.py

from django.db import migrations, models

class Migration(migrations.Migration):
    dependencies = [
        ('app', '0041_previous_migration'),
    ]

    operations = [
        # SAFE: Add nullable field
        migrations.AddField(
            model_name='user',
            name='bio',
            field=models.TextField(null=True, blank=True),
        ),

        # SAFE: Add new table
        migrations.CreateModel(
            name='UserProfile',
            fields=[
                ('id', models.BigAutoField(primary_key=True)),
                ('user', models.OneToOneField('User', on_delete=models.CASCADE)),
                ('avatar_url', models.URLField(null=True)),
            ],
        ),

        # SAFE: Add index (will use CONCURRENT on PostgreSQL)
        migrations.AddIndex(
            model_name='user',
            index=models.Index(fields=['email'], name='user_email_idx'),
        ),
    ]

TypeORM Migrations

// Generated migration: 1706234567890-AddUserProfileFields.ts

import { MigrationInterface, QueryRunner, TableColumn, Table } from "typeorm";

export class AddUserProfileFields1706234567890 implements MigrationInterface {
    name = 'AddUserProfileFields1706234567890'

    public async up(queryRunner: QueryRunner): Promise<void> {
        // SAFE: Add nullable column
        await queryRunner.addColumn('users', new TableColumn({
            name: 'bio',
            type: 'text',
            isNullable: true
        }));

        // SAFE: Create new table
        await queryRunner.createTable(new Table({
            name: 'user_profiles',
            columns: [
                {
                    name: 'id',
                    type: 'uuid',
                    isPrimary: true,
                    generationStrategy: 'uuid',
                    default: 'uuid_generate_v4()'
                },
                {
                    name: 'user_id',
                    type: 'uuid',
                },
                {
                    name: 'avatar_url',
                    type: 'varchar',
                    isNullable: true
                }
            ]
        }));

        // SAFE: Add index concurrently (PostgreSQL)
        await queryRunner.query(
            `CREATE INDEX CONCURRENTLY "idx_users_email" ON "users" ("email")`
        );
    }

    public async down(queryRunner: QueryRunner): Promise<void> {
        // Reverse operations
        await queryRunner.query(`DROP INDEX "idx_users_email"`);
        await queryRunner.dropTable('user_profiles');
        await queryRunner.dropColumn('users', 'bio');
    }
}

Prisma Migrations

-- Migration: 20260125000000_add_user_profile_fields

-- CreateTable (SAFE)
CREATE TABLE "user_profiles" (
    "id" UUID NOT NULL DEFAULT gen_random_uuid(),
    "user_id" UUID NOT NULL,
    "avatar_url" TEXT,
    "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,

    CONSTRAINT "user_profiles_pkey" PRIMARY KEY ("id")
);

-- AddColumn (SAFE - nullable)
ALTER TABLE "users" ADD COLUMN "bio" TEXT;

-- CreateIndex (SAFE - using CONCURRENTLY)
CREATE INDEX CONCURRENTLY "idx_users_email" ON "users"("email");

-- AddForeignKey (SAFE - can be added online)
ALTER TABLE "user_profiles"
ADD CONSTRAINT "user_profiles_user_id_fkey"
FOREIGN KEY ("user_id")
REFERENCES "users"("id")
ON DELETE CASCADE;

SQLAlchemy/​Alembic Migrations

# Generated migration: add_user_profile_fields.py

"""Add user profile fields

Revision ID: abc123def456
Revises: prev123rev456
Create Date: 2026-01-25 10:00:00.000000
"""

from alembic import op
import sqlalchemy as sa

revision = 'abc123def456'
down_revision = 'prev123rev456'
branch_labels = None
depends_on = None

def upgrade():
    # SAFE: Add nullable column
    op.add_column('users', sa.Column('bio', sa.Text(), nullable=True))

    # SAFE: Create new table
    op.create_table('user_profiles',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('user_id', sa.Integer(), nullable=False),
        sa.Column('avatar_url', sa.String(500), nullable=True),
        sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
        sa.PrimaryKeyConstraint('id')
    )

    # SAFE: Add index concurrently (PostgreSQL)
    op.create_index(
        'idx_users_email',
        'users',
        ['email'],
        postgresql_concurrently=True
    )

def downgrade():
    op.drop_index('idx_users_email', table_name='users')
    op.drop_table('user_profiles')
    op.drop_column('users', 'bio')

Phase 4: Complex Migration Strategies

For risky operations, I'll generate multi-step migrations:

Strategy 1: Column Rename (Zero Downtime)

# Step 1: Add new column (deploy with this)
class Migration1(migrations.Migration):
    operations = [
        migrations.AddField(
            model_name='user',
            name='full_name',  # New name
            field=models.CharField(max_length=255, null=True),
        ),
    ]

# Step 2: Backfill data (run after deployment)
class Migration2(migrations.Migration):
    dependencies = [('app', '0001_add_full_name')],

    operations = [
        migrations.RunPython(backfill_full_name, reverse_code=migrations.RunPython.noop),
    ]

def backfill_full_name(apps, schema_editor):
    User = apps.get_model('app', 'User')
    User.objects.filter(full_name__isnull=True).update(
        full_name=models.F('name')  # Copy from old column
    )

# Step 3: Make NOT NULL (after backfill completes)
class Migration3(migrations.Migration):
    operations = [
        migrations.AlterField(
            model_name='user',
            name='full_name',
            field=models.CharField(max_length=255),  # Remove null=True
        ),
    ]

# Step 4: Remove old column (after next deployment)
class Migration4(migrations.Migration):
    operations = [
        migrations.RemoveField(
            model_name='user',
            name='name',  # Old column
        ),
    ]

Strategy 2: Add NOT NULL Column (Zero Downtime)

# Step 1: Add nullable column with default
class Migration1(migrations.Migration):
    operations = [
        migrations.AddField(
            model_name='user',
            name='status',
            field=models.CharField(max_length=20, null=True, default='active'),
        ),
    ]

# Step 2: Backfill existing rows
class Migration2(migrations.Migration):
    operations = [
        migrations.RunPython(backfill_status, reverse_code=migrations.RunPython.noop),
    ]

def backfill_status(apps, schema_editor):
    User = apps.get_model('app', 'User')
    User.objects.filter(status__isnull=True).update(status='active')

# Step 3: Add NOT NULL constraint
class Migration3(migrations.Migration):
    operations = [
        migrations.AlterField(
            model_name='user',
            name='status',
            field=models.CharField(max_length=20, default='active'),  # Remove null=True
        ),
    ]

Strategy 3: Change Column Type (Zero Downtime)

-- Step 1: Add new column with new type
ALTER TABLE users ADD COLUMN age_new INTEGER;

-- Step 2: Backfill data with conversion
UPDATE users SET age_new = age::INTEGER WHERE age_new IS NULL;

-- Step 3: Verify data integrity
SELECT COUNT(*) FROM users WHERE age IS NOT NULL AND age_new IS NULL;
-- Should return 0

-- Step 4: Create index on new column
CREATE INDEX CONCURRENTLY idx_users_age_new ON users(age_new);

-- Step 5: Drop old index
DROP INDEX idx_users_age;

-- Step 6: Rename columns (requires exclusive lock, but very fast)
BEGIN;
ALTER TABLE users RENAME COLUMN age TO age_old;
ALTER TABLE users RENAME COLUMN age_new TO age;
COMMIT;

-- Step 7: Drop old column (in next migration, after deployment)
ALTER TABLE users DROP COLUMN age_old;

Phase 5: Migration Testing

I'll generate test scripts for the migration:

#!/​bin/​bash
# Test migration safety

test_migration() {
    local migration_file="$1"

    echo "=== Testing Migration: $migration_file ==="
    echo ""

    # 1. Test on fresh database
    echo "Test 1: Fresh database migration"
    dropdb test_db_fresh 2>/​dev/​null
    createdb test_db_fresh
    psql test_db_fresh < schema_dump.sql
    python manage.py migrate --database=test_db_fresh

    if [ $? -eq 0 ]; then
        echo "✓ Fresh migration successful"
    else
        echo "❌ Fresh migration failed"
        exit 1
    fi

    # 2. Test rollback
    echo ""
    echo "Test 2: Migration rollback"
    python manage.py migrate app $(get_previous_migration) --database=test_db_fresh

    if [ $? -eq 0 ]; then
        echo "✓ Rollback successful"
    else
        echo "❌ Rollback failed"
        exit 1
    fi

    # 3. Test data preservation
    echo ""
    echo "Test 3: Data preservation"
    psql test_db_fresh -c "SELECT COUNT(*) FROM users;"
    # Verify count hasn't changed

    # 4. Performance test
    echo ""
    echo "Test 4: Migration performance"
    time python manage.py migrate --database=test_db_fresh

    echo ""
    echo "✓ All migration tests passed"
}

test_migration "migrations/​0042_add_user_profile.py"

Phase 6: Migration Documentation

I'll generate comprehensive migration documentation:

# Migration Guide: Add User Profile Fields

**Migration**: `0042_add_user_profile_fields`
**Type**: Schema Addition
**Risk Level**: Low (all operations are safe)
**Estimated Duration**: < 1 minute
**Downtime Required**: None

## Changes

### New Columns
- `users.bio` (TEXT, nullable)
  - Purpose: Store user biography
  - Default: NULL
  - Index: None

### New Tables
- `user_profiles`
  - Columns: id, user_id, avatar_url
  - Indexes: PRIMARY KEY (id)
  - Foreign Keys: user_id → users.id

### New Indexes
- `idx_users_email` on `users(email)`
  - Type: B-tree
  - Created CONCURRENTLY: Yes (no table lock)

## Deployment Steps

### Pre-Deployment
1. Review migration file
2. Test on staging environment
3. Backup production database
4. Schedule deployment window

### Deployment
```bash
# Run migration
python manage.py migrate

# Verify
python manage.py showmigrations app

Post-Deployment

1. Verify new columns exist
2. Check index creation
3. Monitor query performance
4. Verify application functionality

Rollback Plan

If issues occur:

# Rollback migration
python manage.py migrate app 0041_previous_migration

# Verify rollback
python manage.py showmigrations app

Note: Rollback is safe and will not lose data.

Performance Impact

• Index Creation: ~10 seconds per million rows (CONCURRENT)
• Table Creation: Instant (no existing data)
• Column Addition: Instant (nullable columns)

Monitoring

Watch for:

• Increased query time on users table
• Lock contention during index creation
• Application errors related to new fields

FAQ

Q: Will this cause downtime? A: No, all operations are online and non-blocking.

Q: Can I run this during business hours? A: Yes, safe to run anytime.

Q: What if the migration fails halfway? A: The migration is atomic; it will rollback automatically.

## Practical Examples

**Generate Migration:**
```bash
/​migration-generate "add bio field to User model"
/​migration-generate "create UserProfile table"
/​migration-generate            # Auto-detect from model changes

Complex Migrations:

/​migration-generate "rename username to email"
/​migration-generate "change age from string to integer"
/​migration-generate "add NOT NULL constraint to status"

Safety Checklist

Before running migrations:

• Backup database
• Test on staging
• Review generated SQL
• Check rollback works
• Plan for data migration (if needed)
• Schedule deployment window
• Prepare monitoring

What I'll Actually Do

1. Detect changes - Use Grep to find modified schemas
2. Analyze complexity - Determine safe vs risky operations
3. Generate migration - Framework-appropriate migration files
4. Plan strategy - Multi-step for complex changes
5. Create tests - Validation scripts
6. Document - Comprehensive deployment guide

Important: I will NEVER:

• Generate migrations without rollback capability
• Skip data preservation checks
• Recommend unsafe operations without alternatives
• Add AI attribution

All migrations will be production-safe, well-documented, and thoroughly tested.

Credits: Migration safety patterns based on PostgreSQL documentation and Django/​Rails migration best practices.