database-migration-patterns（数据库迁移模式）

Database Migration Patterns

Best practices for safe database schema migrations.

Migration File Structure

migrations/
├── versions/
│   ├── 001_initial_schema.py
│   ├── 002_add_users_table.py
│   ├── 003_add_user_email_index.py
│   └── 004_add_orders_table.py
├── alembic.ini
└── env.py

Alembic Migration Template

"""Add users table

Revision ID: 002
Revises: 001
Create Date: 2024-01-15 10:00:00.000000
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

# revision identifiers
revision = '002'
down_revision = '001'
branch_labels = None
depends_on = None

def upgrade() -> None:
    op.create_table(
        'users',
        sa.Column('id', postgresql.UUID(as_uuid=True), primary_key=True),
        sa.Column('email', sa.String(255), nullable=False),
        sa.Column('name', sa.String(255), nullable=True),
        sa.Column('status', sa.String(50), nullable=False, server_default='pending'),
        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()),
        sa.Column('updated_at', sa.DateTime(timezone=True), onupdate=sa.func.now()),
    )

    op.create_index('ix_users_email', 'users', ['email'], unique=True)
    op.create_index('ix_users_status', 'users', ['status'])

def downgrade() -> None:
    op.drop_index('ix_users_status')
    op.drop_index('ix_users_email')
    op.drop_table('users')

Zero-Downtime Migration Patterns

Pattern 1: Expand and Contract

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

# Step 2: Backfill data (run separately, possibly in batches)
def backfill():
    connection = op.get_bind()
    connection.execute("""
        UPDATE users
        SET email_new = email
        WHERE email_new IS NULL
        LIMIT 10000
    """)

# Step 3: Make new column non-nullable, drop old column
def upgrade_step3():
    op.alter_column('users', 'email_new', nullable=False)
    op.drop_column('users', 'email')
    op.alter_column('users', 'email_new', new_column_name='email')

Pattern 2: Safe Column Rename

# DON'T: Direct rename causes downtime
# op.alter_column('users', 'username', new_column_name='email')

# DO: Expand-Contract pattern
def upgrade():
    # 1. Add new column
    op.add_column('users', sa.Column('email', sa.String(255)))

    # 2. Create trigger to sync data (PostgreSQL)
    op.execute("""
        CREATE OR REPLACE FUNCTION sync_username_to_email()
        RETURNS TRIGGER AS $$
        BEGIN
            NEW.email = NEW.username;
            RETURN NEW;
        END;
        $$ LANGUAGE plpgsql;

        CREATE TRIGGER sync_username_email
        BEFORE INSERT OR UPDATE ON users
        FOR EACH ROW EXECUTE FUNCTION sync_username_to_email();
    """)

    # 3. Backfill existing data
    op.execute("UPDATE users SET email = username WHERE email IS NULL")

# Later migration after app updated:
def upgrade_cleanup():
    op.execute("DROP TRIGGER sync_username_email ON users")
    op.execute("DROP FUNCTION sync_username_to_email()")
    op.drop_column('users', 'username')

Pattern 3: Safe Index Creation

def upgrade():
    # Create index concurrently to avoid locking
    op.execute("""
        CREATE INDEX CONCURRENTLY ix_users_email
        ON users (email)
    """)

def downgrade():
    op.execute("DROP INDEX CONCURRENTLY ix_users_email")

Pattern 4: Adding NOT NULL Constraint

def upgrade():
    # 1. Add column as nullable
    op.add_column('users', sa.Column('verified', sa.Boolean(), nullable=True))

    # 2. Set default for new rows
    op.alter_column('users', 'verified', server_default=sa.false())

    # 3. Backfill existing rows (in batches for large tables)
    op.execute("""
        UPDATE users
        SET verified = false
        WHERE verified IS NULL
    """)

    # 4. Add NOT NULL constraint
    op.alter_column('users', 'verified', nullable=False)

Batch Processing for Large Tables

from sqlalchemy import text

def backfill_in_batches(connection, batch_size=10000):
    """Backfill data in batches to avoid long locks."""
    while True:
        result = connection.execute(text("""
            UPDATE users
            SET new_column = old_column
            WHERE id IN (
                SELECT id FROM users
                WHERE new_column IS NULL
                LIMIT :batch_size
                FOR UPDATE SKIP LOCKED
            )
            RETURNING id
        """), {"batch_size": batch_size})

        updated = result.rowcount
        connection.commit()

        if updated == 0:
            break

        print(f"Updated {updated} rows")
        time.sleep(0.1)  # Small delay to reduce load

Data Migration Patterns

Separate Data Migrations

# Schema migration (runs during deploy)
def upgrade():
    op.add_column('orders', sa.Column('total_cents', sa.BigInteger()))

# Data migration (runs separately)
# data_migrations/migrate_order_totals.py
def run_data_migration():
    """Convert total from dollars to cents."""
    with engine.connect() as conn:
        while True:
            result = conn.execute(text("""
                UPDATE orders
                SET total_cents = total * 100
                WHERE total_cents IS NULL
                AND id IN (
                    SELECT id FROM orders
                    WHERE total_cents IS NULL
                    LIMIT 5000
                )
            """))

            if result.rowcount == 0:
                break

            conn.commit()

Foreign Key Constraints

Safe Foreign Key Addition

def upgrade():
    # 1. Add column without constraint
    op.add_column('orders',
        sa.Column('user_id', postgresql.UUID(), nullable=True)
    )

    # 2. Backfill data
    op.execute("""
        UPDATE orders o
        SET user_id = (
            SELECT id FROM users u
            WHERE u.legacy_id = o.legacy_user_id
        )
    """)

    # 3. Add constraint with NOT VALID (PostgreSQL)
    op.execute("""
        ALTER TABLE orders
        ADD CONSTRAINT fk_orders_user_id
        FOREIGN KEY (user_id) REFERENCES users(id)
        NOT VALID
    """)

    # 4. Validate constraint in background
    op.execute("""
        ALTER TABLE orders
        VALIDATE CONSTRAINT fk_orders_user_id
    """)

Version Control Integration

# CI/CD pipeline for migrations
name: Database Migration

on:
  push:
    paths:
      - 'migrations/**'

jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Validate migration
        run: |
          alembic check
          alembic upgrade head --sql > /dev/null

      - name: Check for destructive operations
        run: |
          # Fail if migration contains DROP without review
          if grep -r "op.drop" migrations/versions/*.py; then
            echo "::warning::Migration contains DROP operations"
          fi

  deploy-staging:
    needs: validate
    runs-on: ubuntu-latest
    environment: staging
    steps:
      - name: Run migration
        run: alembic upgrade head

Rollback Strategies

# Always implement downgrade
def downgrade():
    # For additive changes, downgrade is straightforward
    op.drop_column('users', 'new_column')

# For destructive changes, preserve data
def upgrade():
    # Rename instead of drop
    op.rename_table('old_table', '_old_table_backup')

def downgrade():
    op.rename_table('_old_table_backup', 'old_table')

Testing Migrations

# tests/test_migrations.py
import pytest
from alembic.config import Config
from alembic import command

@pytest.fixture
def alembic_config():
    config = Config("alembic.ini")
    return config

def test_upgrade_downgrade(alembic_config, test_database):
    """Test that all migrations can upgrade and downgrade."""
    # Upgrade to head
    command.upgrade(alembic_config, "head")

    # Downgrade to base
    command.downgrade(alembic_config, "base")

    # Upgrade again
    command.upgrade(alembic_config, "head")

def test_migration_is_reversible(alembic_config, test_database):
    """Test each migration individually."""
    revisions = get_all_revisions()

    for rev in revisions:
        command.upgrade(alembic_config, rev)
        command.downgrade(alembic_config, "-1")
        command.upgrade(alembic_config, rev)

References

• Alembic Documentation
• PostgreSQL ALTER TABLE
• Zero-Downtime Migrations
• Safe Database Migrations