09 - Migraciones con Alembic

1. Alembic vs Drizzle Kit

// Drizzle Kit
// 1. Generar migración
npx drizzle-kit generate:pg

// 2. Aplicar
npx drizzle-kit push:pg

// Genera SQL puro, sin lógica de datos

# Alembic
# 1. Generar migración
alembic revision --autogenerate -m "add users table"

# 2. Aplicar
alembic upgrade head

# Genera Python, permite data migrations

Ventaja clave de Alembic: Las migraciones son código Python, no solo SQL. Puedes ejecutar lógica compleja durante la migración (transformar datos, backfill, etc.).

2. Setup Inicial

# Instalar
uv add alembic

# Inicializar (crea carpeta alembic/)
alembic init alembic

Estructura generada

alembic/
├── env.py              # Configuración del entorno
├── script.py.mako      # Template para nuevas migraciones
├── versions/           # Archivos de migración
│   └── .gitkeep
alembic.ini             # Configuración principal

Configurar para Async

# alembic/env.py
import asyncio
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context

from app.database import Base  # Tu Base declarativa
from app.core.config import settings

config = context.config

# Configurar logging
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# Metadata de tus modelos
target_metadata = Base.metadata

def get_url():
    return settings.DATABASE_URL

def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode (genera SQL sin conectar)."""
    url = get_url()
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()

def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)

    with context.begin_transaction():
        context.run_migrations()

async def run_async_migrations() -> None:
    """Run migrations in 'online' mode with async engine."""
    configuration = config.get_section(config.config_ini_section)
    configuration["sqlalchemy.url"] = get_url()
    
    connectable = async_engine_from_config(
        configuration,
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()

def run_migrations_online() -> None:
    """Run migrations in 'online' mode."""
    asyncio.run(run_async_migrations())

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

alembic.ini

[alembic]
script_location = alembic
prepend_sys_path = .
version_path_separator = os

# La URL se configura en env.py
sqlalchemy.url = 

[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARN
handlers = console

[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
handlers =
qualname = alembic

3. Workflow de Migraciones

Crear Migración Autogenerada

# Detecta cambios entre modelos y DB
alembic revision --autogenerate -m "add posts table"

Migración Generada

# alembic/versions/xxxx_add_posts_table.py
"""add posts table

Revision ID: a1b2c3d4e5f6
Revises: 9z8y7x6w5v4u
Create Date: 2024-02-01 10:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa

revision: str = 'a1b2c3d4e5f6'
down_revision: Union[str, None] = '9z8y7x6w5v4u'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None

def upgrade() -> None:
    op.create_table(
        'posts',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('title', sa.String(length=200), nullable=False),
        sa.Column('content', sa.Text(), nullable=False),
        sa.Column('author_id', sa.Integer(), nullable=False),
        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
        sa.ForeignKeyConstraint(['author_id'], ['users.id'], ),
        sa.PrimaryKeyConstraint('id')
    )
    op.create_index(op.f('ix_posts_author_id'), 'posts', ['author_id'], unique=False)

def downgrade() -> None:
    op.drop_index(op.f('ix_posts_author_id'), table_name='posts')
    op.drop_table('posts')

Comandos Esenciales

# Ver estado actual
alembic current

# Ver historial
alembic history --verbose

# Aplicar todas las migraciones pendientes
alembic upgrade head

# Aplicar una migración específica
alembic upgrade a1b2c3d4e5f6

# Rollback una migración
alembic downgrade -1

# Rollback a versión específica
alembic downgrade 9z8y7x6w5v4u

# Generar SQL sin aplicar (para review)
alembic upgrade head --sql > migration.sql

4. Data Migrations

La killer feature de Alembic: ejecutar código Python durante migraciones.

Ejemplo: Backfill de Campo Calculado

"""add full_name column and backfill

Revision ID: b2c3d4e5f6g7
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.orm import Session

def upgrade() -> None:
    # 1. Añadir columna (nullable inicialmente)
    op.add_column('users', sa.Column('full_name', sa.String(200), nullable=True))
    
    # 2. Backfill con datos existentes
    bind = op.get_bind()
    session = Session(bind=bind)
    
    # Usar SQL directo para eficiencia
    session.execute(
        sa.text("""
            UPDATE users 
            SET full_name = first_name || ' ' || last_name
            WHERE full_name IS NULL
        """)
    )
    session.commit()
    
    # 3. Hacer NOT NULL después del backfill
    op.alter_column('users', 'full_name', nullable=False)

def downgrade() -> None:
    op.drop_column('users', 'full_name')

Ejemplo: Migración de Datos Entre Tablas

"""split address into separate table

Revision ID: c3d4e5f6g7h8
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.orm import Session

def upgrade() -> None:
    # 1. Crear nueva tabla
    op.create_table(
        'addresses',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('user_id', sa.Integer(), sa.ForeignKey('users.id'), nullable=False),
        sa.Column('street', sa.String(200)),
        sa.Column('city', sa.String(100)),
        sa.Column('postal_code', sa.String(20)),
    )
    
    # 2. Migrar datos
    bind = op.get_bind()
    session = Session(bind=bind)
    
    session.execute(sa.text("""
        INSERT INTO addresses (user_id, street, city, postal_code)
        SELECT id, address_street, address_city, address_postal
        FROM users
        WHERE address_street IS NOT NULL
    """))
    session.commit()
    
    # 3. Eliminar columnas viejas
    op.drop_column('users', 'address_street')
    op.drop_column('users', 'address_city')
    op.drop_column('users', 'address_postal')

def downgrade() -> None:
    # Restaurar columnas
    op.add_column('users', sa.Column('address_street', sa.String(200)))
    op.add_column('users', sa.Column('address_city', sa.String(100)))
    op.add_column('users', sa.Column('address_postal', sa.String(20)))
    
    # Migrar datos de vuelta
    bind = op.get_bind()
    session = Session(bind=bind)
    
    session.execute(sa.text("""
        UPDATE users u
        SET 
            address_street = a.street,
            address_city = a.city,
            address_postal = a.postal_code
        FROM addresses a
        WHERE a.user_id = u.id
    """))
    session.commit()
    
    op.drop_table('addresses')

5. Migraciones Seguras en Producción

Reglas de Oro

1. Nunca hagas DROP COLUMN directamente — Primero haz la columna nullable, despliega código que no la use, luego elimina
2. Añade columnas como nullable — Luego backfill, luego NOT NULL
3. Índices CONCURRENTLY — En PostgreSQL, crear índices sin bloquear

Crear Índice Sin Bloqueo

from alembic import op

def upgrade() -> None:
    # Usar execute() para SQL raw con CONCURRENTLY
    op.execute("""
        CREATE INDEX CONCURRENTLY ix_posts_created_at 
        ON posts (created_at)
    """)

def downgrade() -> None:
    op.execute("DROP INDEX CONCURRENTLY ix_posts_created_at")

Rename de Columna Seguro (Zero Downtime)

# Migración 1: Añadir nueva columna
def upgrade() -> None:
    op.add_column('users', sa.Column('email_address', sa.String(255)))
    # Trigger para sincronizar durante transición
    op.execute("""
        CREATE OR REPLACE FUNCTION sync_email()
        RETURNS TRIGGER AS $$
        BEGIN
            NEW.email_address := NEW.email;
            RETURN NEW;
        END;
        $$ LANGUAGE plpgsql;
        
        CREATE TRIGGER sync_email_trigger
        BEFORE INSERT OR UPDATE ON users
        FOR EACH ROW EXECUTE FUNCTION sync_email();
    """)
    # Backfill existentes
    op.execute("UPDATE users SET email_address = email")

# Migración 2 (después de desplegar código que use email_address):
def upgrade() -> None:
    op.drop_column('users', 'email')
    op.execute("DROP TRIGGER sync_email_trigger ON users")
    op.execute("DROP FUNCTION sync_email()")

6. Testing de Migraciones

# tests/test_migrations.py
import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine, text

@pytest.fixture
def alembic_config():
    config = Config("alembic.ini")
    config.set_main_option("sqlalchemy.url", "postgresql://test:test@localhost/test_db")
    return config

def test_migrations_up_down(alembic_config):
    """Test que todas las migraciones aplican y revierten correctamente."""
    # Upgrade all
    command.upgrade(alembic_config, "head")
    
    # Downgrade all
    command.downgrade(alembic_config, "base")
    
    # Upgrade again (verifica idempotencia)
    command.upgrade(alembic_config, "head")

def test_migration_generates_expected_tables(alembic_config):
    """Test que las migraciones crean las tablas esperadas."""
    command.upgrade(alembic_config, "head")
    
    engine = create_engine(alembic_config.get_main_option("sqlalchemy.url"))
    with engine.connect() as conn:
        result = conn.execute(text("""
            SELECT table_name FROM information_schema.tables 
            WHERE table_schema = 'public'
        """))
        tables = {row[0] for row in result}
    
    assert "users" in tables
    assert "posts" in tables
    assert "alembic_version" in tables

7. CI/CD con Migraciones

GitHub Actions

# .github/workflows/migrate.yml
name: Database Migrations

on:
  push:
    branches: [main]
    paths:
      - 'alembic/**'

jobs:
  migrate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      
      - name: Install dependencies
        run: |
          pip install uv
          uv sync --frozen
      
      - name: Run migrations
        env:
          DATABASE_URL: ${{ secrets.DATABASE_URL }}
        run: |
          uv run alembic upgrade head

Script de Despliegue

#!/bin/bash
# deploy.sh

set -e

echo "Running database migrations..."
alembic upgrade head

echo "Starting application..."
exec gunicorn -k uvicorn.workers.UvicornWorker main:app

Docker Entrypoint

# Dockerfile
COPY --chmod=755 scripts/entrypoint.sh /entrypoint.sh
ENTRYPOINT ["/entrypoint.sh"]
CMD ["gunicorn", "-k", "uvicorn.workers.UvicornWorker", "main:app"]

#!/bin/bash
# scripts/entrypoint.sh

set -e

# Esperar a que la DB esté lista
echo "Waiting for database..."
while ! pg_isready -h $DB_HOST -p $DB_PORT -U $DB_USER; do
  sleep 1
done

# Aplicar migraciones
echo "Running migrations..."
alembic upgrade head

# Ejecutar comando principal
exec "$@"

8. Múltiples Bases de Datos

# alembic/env.py para múltiples DBs
from alembic import context

def run_migrations_online():
    # Determinar qué DB migrar basándose en argumento
    db_name = context.get_x_argument(as_dictionary=True).get('db', 'main')
    
    if db_name == 'main':
        url = settings.MAIN_DATABASE_URL
        metadata = main_metadata
    elif db_name == 'analytics':
        url = settings.ANALYTICS_DATABASE_URL
        metadata = analytics_metadata
    
    # ... resto de configuración

# Uso
alembic -x db=main upgrade head
alembic -x db=analytics upgrade head

9. Troubleshooting Común

”Target database is not up to date”

# Ver estado actual
alembic current

# Marcar como aplicada sin ejecutar (si ya aplicaste manualmente)
alembic stamp head

Conflictos de Branches

# Ver branches
alembic branches

# Merge branches
alembic merge -m "merge branches" rev1 rev2

Migración Autogenerate No Detecta Cambios

Asegúrate de importar todos los modelos en env.py:

# alembic/env.py
from app.models.user import User  # Importar explícitamente
from app.models.post import Post
from app.database import Base

target_metadata = Base.metadata

10. Comparativa Final

Conclusión

Alembic ofrece un control superior sobre migraciones comparado con herramientas de Node.js:

1. Migraciones como código — Python completo, no solo SQL
2. Data migrations — Transforma datos durante schema changes
3. Downgrades — Rollback estructurado
4. Branches — Maneja desarrollo paralelo

Pattern Senior: Siempre revisa las migraciones autogeneradas antes de aplicar. Alembic puede generar operaciones destructivas si no entiende tu intención.

En el siguiente capítulo, implementaremos autenticación y autorización con OAuth2 y JWT.