Database Migrations & Schema Evolution

# Install Alembic
pip install alembic sqlalchemy psycopg2-binary

# Initialize Alembic in your project
alembic init migrations

# This creates:
# migrations/
#   ├── alembic.ini          # Configuration file
#   ├── env.py               # Environment setup
#   ├── script.py.mako       # Migration template
#   └── versions/            # Migration files go here

# alembic.ini - Update database URL
sqlalchemy.url = postgresql://postgres:password@localhost/mydb

# Or use environment variable (recommended for production)
# sqlalchemy.url = driver://user:pass@localhost/dbname

# migrations/env.py - Configure SQLAlchemy models
from myapp.models import Base  # Your SQLAlchemy Base

target_metadata = Base.metadata  # Auto-detect model changes

# Define SQLAlchemy model
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

# Auto-generate migration from model
alembic revision --autogenerate -m "Create users table"

# This creates: migrations/versions/abc123_create_users_table.py

# migrations/versions/abc123_create_users_table.py
"""Create users table

Revision ID: abc123
Revises:
Create Date: 2025-02-05 14:30:22.123456
"""
from alembic import op
import sqlalchemy as sa

# Revision identifiers
revision = 'abc123'
down_revision = None  # First migration
branch_labels = None
depends_on = None

def upgrade():
    """Apply the migration (forward)"""
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('username', sa.String(length=50), nullable=False),
        sa.Column('email', sa.String(length=100), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=True),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('username')
    )

def downgrade():
    """Rollback the migration (backward)"""
    op.drop_table('users')

# Apply all pending migrations
alembic upgrade head

# Output shows:
# INFO  [alembic.runtime.migration] Running upgrade -> abc123, Create users table
# INFO  [alembic.runtime.migration] Migration complete

# Check current version
alembic current

# Output:
# abc123 (head)

# Phase 1: Add column (nullable, no default = instant)
def upgrade():
    # Safe: nullable column with no default = no table rewrite
    op.add_column('users', sa.Column('full_name', sa.String(200), nullable=True))

    # Alembic command:
    alembic revision -m "Phase 1: Add full_name column"
    alembic upgrade head

# Deploy code that populates full_name when creating/updating users
class User(Base):
    __tablename__ = 'users'
    username = Column(String(50))
    full_name = Column(String(200), nullable=True)  # New column

    @property
    def name_for_display(self):
        return self.full_name or self.username  # Fallback to old

# Backfill script (run in batches to avoid locking)
from sqlalchemy.orm import Session
from myapp.models import User

def backfill_full_names(batch_size=1000):
    """Populate full_name for existing users"""
    session = Session()

    # Get users without full_name
    total = 0
    while True:
        users = session.query(User).filter(
            User.full_name == None
        ).limit(batch_size).all()

        if not users:
            break  # Done!

        # Update in batches
        for user in users:
            user.full_name = f"{user.first_name} {user.last_name}"

        session.commit()
        total += len(users)
        print(f"Backfilled {total} users...")

    print("Backfill complete!")

# Run backfill (can run during business hours, no locks)
backfill_full_names()

# Phase 3: Add NOT NULL constraint (safe because all rows have values)
def upgrade():
    # Add NOT NULL constraint
    # PostgreSQL 11+: Use NOT VALID then VALIDATE for zero downtime
    op.execute('ALTER TABLE users ALTER COLUMN full_name SET NOT NULL')

# Alembic command:
alembic revision -m "Phase 3: Make full_name NOT NULL"
alembic upgrade head

# Now all new code can assume full_name always exists!

# Check current version
alembic current
# Output: def456 (head)

# Rollback to previous version
alembic downgrade -1

# Output:
# INFO  [alembic.runtime.migration] Running downgrade def456 -> abc123
# INFO  [alembic.runtime.migration] Rollback complete

# Or rollback to specific version
alembic downgrade abc123

# Or rollback everything
alembic downgrade base

# .github/workflows/deploy.yml
name: Deploy with Migrations

on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2

      - name: Setup Python
        uses: actions/setup-python@v2

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Check migration files
        run: |
          # Ensure migrations are valid
          alembic check

      - name: Apply migrations (staging)
        run: |
          alembic upgrade head
        env:
          DATABASE_URL: ${{ secrets.STAGING_DB_URL }}

      - name: Run tests
        run: pytest

      - name: Apply migrations (production)
        if: success()
        run: |
          alembic upgrade head
        env:
          DATABASE_URL: ${{ secrets.PROD_DB_URL }}

      - name: Deploy application
        run: |
          # Deploy code after migrations succeed
          ./deploy.sh