Migrações com Alembic

1. Introdução ao Alembic

Alembic é uma ferramenta de migração de banco de dados leve e poderosa para Python, desenvolvida pelo mesmo criador do SQLAlchemy. Ela permite versionar e gerenciar alterações no esquema do banco de dados de forma controlada e reproduzível. Em vez de modificar manualmente tabelas e colunas, você cria "migrações" — arquivos que descrevem as alterações — e as aplica sequencialmente.

A integração com SQLAlchemy é nativa: o Alembic pode inspecionar seus modelos (seja via Core ou ORM) e gerar automaticamente o código de migração correspondente. Isso elimina a necessidade de escrever SQL manual para tarefas rotineiras.

Para instalar:

pip install alembic

Após a instalação, inicialize um ambiente de migração no diretório do seu projeto:

alembic init alembic

Esse comando cria a estrutura básica que exploraremos a seguir.

2. Configuração do Ambiente de Migração

O comando alembic init gera a seguinte estrutura:

projeto/
├── alembic/
│   ├── versions/        # Diretório para os arquivos de migração
│   ├── env.py           # Configuração do ambiente de migração
│   └── script.py.mako   # Template para novas migrações
└── alembic.ini          # Configuração principal

O primeiro passo é configurar a string de conexão com o banco de dados no alembic.ini:

sqlalchemy.url = postgresql://usuario:senha@localhost:5432/meubanco

Em seguida, no arquivo env.py, precisamos vincular nossos modelos SQLAlchemy para que o Alembic possa detectar alterações automaticamente. Supondo que você tenha um arquivo models.py com sua Base declarativa:

from models import Base
target_metadata = Base.metadata

Substitua a linha target_metadata = None pela linha acima. Agora o Alembic sabe quais tabelas e colunas existem e pode compará-las com o estado atual do banco.

3. Criando a Primeira Migração

Com tudo configurado, crie sua primeira migração automática:

alembic revision --autogenerate -m "criar tabela usuario"

O Alembic compara o metadata dos seus modelos com o banco de dados atual e gera um arquivo em versions/. Vamos supor que você tenha o seguinte modelo:

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Usuario(Base):
    __tablename__ = 'usuarios'
    id = Column(Integer, primary_key=True)
    nome = Column(String(100), nullable=False)
    email = Column(String(200), unique=True)

A migração gerada será algo como:

"""criar tabela usuario

Revision ID: a1b2c3d4e5f6
Revises: 
Create Date: 2025-03-20 10:00:00.000000
"""
from alembic import op
import sqlalchemy as sa

revision = 'a1b2c3d4e5f6'
down_revision = None
branch_labels = None
depends_on = None

def upgrade():
    op.create_table('usuarios',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('nome', sa.String(length=100), nullable=False),
        sa.Column('email', sa.String(length=200), nullable=True),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email')
    )

def downgrade():
    op.drop_table('usuarios')

Aplique a migração com:

alembic upgrade head

O banco agora reflete exatamente seus modelos.

4. Operações Comuns em Migrações

Vamos explorar operações típicas que você encontrará no dia a dia.

Adicionar uma coluna:

def upgrade():
    op.add_column('usuarios', sa.Column('idade', sa.Integer(), nullable=True))

def downgrade():
    op.drop_column('usuarios', 'idade')

Alterar tipo de coluna:

def upgrade():
    op.alter_column('usuarios', 'nome',
                    type_=sa.String(150),
                    nullable=False)

def downgrade():
    op.alter_column('usuarios', 'nome',
                    type_=sa.String(100),
                    nullable=False)

Criar índice e chave estrangeira:

def upgrade():
    op.create_index('idx_usuario_email', 'usuarios', ['email'])
    op.create_foreign_key('fk_pedido_usuario', 'pedidos', 'usuarios',
                          ['usuario_id'], ['id'])

def downgrade():
    op.drop_constraint('fk_pedido_usuario', 'pedidos', type_='foreignkey')
    op.drop_index('idx_usuario_email', table_name='usuarios')

5. Migrações Manuais e Personalizadas

Nem toda migração pode ser gerada automaticamente. Operações como renomear tabelas, migrar dados ou executar SQL complexo exigem escrita manual.

Para criar uma migração em branco:

alembic revision -m "migrar dados de email"

Executar SQL arbitrário:

def upgrade():
    op.execute("""
        UPDATE usuarios
        SET email = LOWER(email)
        WHERE email IS NOT NULL
    """)

def downgrade():
    # Não é possível reverter a transformação de dados facilmente
    pass

Migração de dados entre tabelas:

def upgrade():
    op.execute("""
        INSERT INTO usuarios_novos (id, nome, email)
        SELECT id, nome, email FROM usuarios_antigos
    """)
    op.drop_table('usuarios_antigos')

def downgrade():
    op.create_table('usuarios_antigos', ...)
    op.execute("""
        INSERT INTO usuarios_antigos (id, nome, email)
        SELECT id, nome, email FROM usuarios_novos
    """)

Nesse caso, o downgrade precisa recriar a tabela antiga e repopular os dados.

6. Versionamento e Controle de Fluxo

O Alembic mantém um histórico completo de versões. Comandos úteis:

# Ver versão atual
alembic current

# Ver histórico completo
alembic history

# Avançar ou retroceder um número específico de versões
alembic upgrade +2
alembic downgrade -1

Trabalhando com branches: Se dois desenvolvedores criarem migrações a partir da mesma base, o Alembic cria um branch. Para resolver:

alembic merge -m "merge branches" <revision1> <revision2>

Resolução de conflitos: Em equipes, é comum que migrações concorrentes entrem em conflito. A melhor prática é sempre usar alembic upgrade head antes de criar uma nova migração, garantindo que você está na versão mais recente.

7. Boas Práticas e Troubleshooting

Migrações idempotentes: Sempre escreva migrações que possam ser executadas múltiplas vezes sem erro. Use verificações como:

def upgrade():
    conn = op.get_bind()
    inspector = inspect(conn)
    if 'coluna_temp' not in [c['name'] for c in inspector.get_columns('usuarios')]:
        op.add_column('usuarios', sa.Column('coluna_temp', sa.Integer()))

Testando migrações: Crie um banco de testes e execute:

alembic upgrade head --sql > migracao.sql

Isso gera o SQL sem aplicar ao banco, permitindo revisão manual.

Erros comuns e soluções:

  • Tabela já existe: Use op.create_table(..., if_not_exists=True) (disponível em versões recentes) ou envolva em try/except.
  • Coluna faltante: Verifique se a migração anterior foi aplicada com alembic current.
  • Erro de chave estrangeira: Garanta a ordem correta de criação/drop das tabelas.

Debug com --sql: Para depurar, gere o SQL sem executar:

alembic upgrade head --sql

Isso mostra exatamente quais comandos serão executados, facilitando a identificação de problemas.

Referências