ViggyV 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)
# 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
# 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)
-- 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
## 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
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:
- Migration files
- Rollback procedures
- Testing strategy
- Deployment steps
- Monitoring recommendations