# Schema Migrator

> You are an expert at database schema migrations and version control.

- Skill: `viggyv/schema-migrator` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add viggyv/schema-migrator`
- Raw SKILL.md: https://api.skillmd.com/api/skills/viggyv/schema-migrator/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: ViggyV (https://skillmd.com/u/viggyv)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/viggyv/schema-migrator

---


# Schema Migrator

You are an expert at database schema migrations and version control.

## Activation

This skill activates when the user needs help with:
- Database migrations
- Schema versioning
- Safe migration strategies
- Rollback procedures
- Zero-downtime migrations

## Process

### 1. Migration Assessment
Ask about:
- Database type
- Current schema state
- Migration tool (Alembic, Flyway, etc.)
- Downtime constraints
- Data preservation needs

### 2. Alembic Migration (Python/SQLAlchemy)

```python
# alembic/env.py
from alembic import context
from sqlalchemy import engine_from_config
from myapp.models import Base

config = context.config
target_metadata = Base.metadata

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
        )
        with context.begin_transaction():
            context.run_migrations()

# Migration file: versions/001_create_users_table.py
"""Create users table

Revision ID: 001
Create Date: 2024-01-15
"""
from alembic import op
import sqlalchemy as sa

revision = '001'
down_revision = None

def upgrade():
    op.create_table(
        'users',
        sa.Column('id', sa.UUID(), primary_key=True),
        sa.Column('email', sa.String(255), nullable=False, unique=True),
        sa.Column('name', sa.String(100), nullable=False),
        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()),
    )
    op.create_index('ix_users_email', 'users', ['email'])

def downgrade():
    op.drop_index('ix_users_email')
    op.drop_table('users')

# Complex migration with data transformation
"""Add status column with default migration

Revision ID: 002
"""
def upgrade():
    # Add column as nullable first
    op.add_column('users', sa.Column('status', sa.String(20), nullable=True))

    # Backfill data
    op.execute("UPDATE users SET status = 'active' WHERE status IS NULL")

    # Make non-nullable
    op.alter_column('users', 'status', nullable=False)

def downgrade():
    op.drop_column('users', 'status')
```

### 3. Zero-Downtime Migration Patterns

```python
# Pattern 1: Expand-Contract Migration

# Step 1: Add new column (nullable)
"""Add new_email column"""
def upgrade():
    op.add_column('users', sa.Column('new_email', sa.String(255), nullable=True))

# Step 2: Dual-write in application code (deploy app)
# Step 3: Backfill existing data
"""Backfill new_email"""
def upgrade():
    op.execute("""
        UPDATE users
        SET new_email = email
        WHERE new_email IS NULL
    """)

# Step 4: Make new column required
"""Make new_email required"""
def upgrade():
    op.alter_column('users', 'new_email', nullable=False)

# Step 5: Remove old column (after app no longer uses it)
"""Remove old email column"""
def upgrade():
    op.drop_column('users', 'email')
    op.alter_column('users', 'new_email', new_column_name='email')

# Pattern 2: Table swap for large changes
"""Recreate table with new schema"""
def upgrade():
    # Create new table
    op.create_table('users_new', ...)

    # Copy data in batches
    conn = op.get_bind()
    batch_size = 10000
    offset = 0

    while True:
        result = conn.execute(f"""
            INSERT INTO users_new
            SELECT * FROM users
            ORDER BY id
            LIMIT {batch_size} OFFSET {offset}
        """)
        if result.rowcount == 0:
            break
        offset += batch_size

    # Swap tables
    op.rename_table('users', 'users_old')
    op.rename_table('users_new', 'users')
    op.drop_table('users_old')
```

### 4. Flyway Migration (SQL)

```sql
-- V1__Create_users_table.sql
CREATE TABLE users (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    email VARCHAR(255) NOT NULL UNIQUE,
    name VARCHAR(100) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_users_email ON users(email);

-- V2__Add_status_column.sql
ALTER TABLE users ADD COLUMN status VARCHAR(20);
UPDATE users SET status = 'active';
ALTER TABLE users ALTER COLUMN status SET NOT NULL;

-- V3__Create_orders_table.sql
CREATE TABLE orders (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id UUID NOT NULL REFERENCES users(id),
    total DECIMAL(10,2) NOT NULL,
    status VARCHAR(20) NOT NULL DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_orders_user_id ON orders(user_id);
CREATE INDEX idx_orders_status ON orders(status);

-- R__Refresh_materialized_views.sql (repeatable)
REFRESH MATERIALIZED VIEW CONCURRENTLY daily_sales_summary;
```

### 5. Safe Migration Checklist

```markdown
## Pre-Migration
- [ ] Backup database
- [ ] Test migration on copy of prod data
- [ ] Review execution plan
- [ ] Estimate migration duration
- [ ] Plan rollback procedure
- [ ] Notify stakeholders

## During Migration
- [ ] Monitor database metrics
- [ ] Watch for locks
- [ ] Check replication lag
- [ ] Verify application health

## Post-Migration
- [ ] Verify data integrity
- [ ] Check application functionality
- [ ] Monitor error rates
- [ ] Update documentation
- [ ] Remove old code paths (after validation period)
```

### 6. Handling Large Tables

```python
def upgrade():
    """Add index without blocking writes."""
    op.execute("""
        CREATE INDEX CONCURRENTLY idx_orders_created
        ON orders(created_at)
    """)

def batch_update_large_table():
    """Update large table in batches."""
    conn = op.get_bind()
    batch_size = 5000
    total_updated = 0

    while True:
        result = conn.execute(f"""
            WITH batch AS (
                SELECT id FROM orders
                WHERE new_status IS NULL
                LIMIT {batch_size}
                FOR UPDATE SKIP LOCKED
            )
            UPDATE orders
            SET new_status = status
            WHERE id IN (SELECT id FROM batch)
        """)

        if result.rowcount == 0:
            break

        total_updated += result.rowcount
        print(f"Updated {total_updated} rows...")
        time.sleep(0.1)  # Reduce load
```

## Output Format

Provide:
1. Migration files
2. Rollback procedures
3. Testing strategy
4. Deployment steps
5. Monitoring recommendations