Pular para o conteúdo

    Capítulo 57, Backend

    Alembic: migrações de banco

    O esquema do banco evolui junto com o código. A migração é o histórico versionado dessa evolução, aplicável em qualquer ambiente de forma repetível.

    Os arquivos deste capítulo estão em exemplos/api_pedidos/.

    O problema

    Você adiciona uma coluna ao modelo. O seu banco local não tem essa coluna, o de teste tampouco, e o de produção ainda menos. Alterar à mão em cada ambiente é erro certo. O Alembic guarda cada mudança como um arquivo de migração versionado, que sabe subir (upgrade) e descer (downgrade) o esquema, e registra em uma tabela do próprio banco qual versão está aplicada.

    Configuração

    O alembic.ini aponta para a pasta das migrações. O env.py é o coração: ele conecta o Alembic aos seus modelos e à mesma configuração de banco que a aplicação usa, para que migração e API nunca divirjam de URL:

    exemplos/api_pedidos/alembic.ini
    [alembic]
script_location = %(here)s/migrations
prepend_sys_path = src
path_separator = os
file_template = %%(rev)s_%%(slug)s

[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARNING
handlers = console
qualname =

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

[logger_alembic]
level = INFO
handlers =
qualname = alembic

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S
    exemplos/api_pedidos/migrations/env.py
    from logging.config import fileConfig

from alembic import context
from sqlalchemy import engine_from_config, pool

from api_pedidos.config import obter_configuracao
from api_pedidos.modelos import Base

config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name, disable_existing_loggers=False)

# A URL vem da configuração da aplicação, a mesma que a API usa. O "%" precisa ser escapado
# porque o ConfigParser do Alembic trata "%" como caractere especial.
url = obter_configuracao().database_url.get_secret_value()
config.set_main_option("sqlalchemy.url", url.replace("%", "%%"))

target_metadata = Base.metadata


def run_migrations_offline() -> None:
    """Gera o SQL sem conectar ao banco (alembic upgrade head --sql)."""
    context.configure(
        url=config.get_main_option("sqlalchemy.url"),
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as conexao:
        context.configure(connection=conexao, target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()


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

    O target_metadata = Base.metadata é o que permite o --autogenerate comparar os seus modelos com o banco real.

    Criar uma migração

    O --autogenerate compara os modelos com o banco e escreve a migração por você. Ele só gera o rascunho, e revisar é obrigatório:

    Terminal
    uv run alembic revision --autogenerate -m "criar pedidos" --rev-id 0001
    Saída
    INFO  [alembic.autogenerate.compare.tables] Detected added table 'pedidos'
INFO  [alembic.autogenerate.compare.constraints] Detected added index 'ix_pedidos_cliente' on '('cliente',)'
INFO  [alembic.autogenerate.compare.tables] Detected added table 'itens_pedido'
Generating migrations/versions/0001_criar_pedidos.py ...  done

    O arquivo gerado, depois da minha revisão (o Alembic escreve comentários e tipagem antiga que eu limpo):

    exemplos/api_pedidos/migrations/versions/0001_criar_pedidos.py
    """criar pedidos

Revision ID: 0001
Revises:
Create Date: 2026-10-06 15:46:53

"""

import sqlalchemy as sa
from alembic import op

revision: str = "0001"
down_revision: str | None = None
branch_labels: str | None = None
depends_on: str | None = None


def upgrade() -> None:
    op.create_table(
        "pedidos",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("cliente", sa.String(length=120), nullable=False),
        sa.Column("status", sa.String(length=20), nullable=False),
        sa.Column("chave_idempotencia", sa.String(length=80), nullable=True),
        sa.Column("criado_em", sa.DateTime(timezone=True), nullable=False),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("chave_idempotencia"),
    )
    op.create_index("ix_pedidos_cliente", "pedidos", ["cliente"], unique=False)
    op.create_table(
        "itens_pedido",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("pedido_id", sa.Integer(), nullable=False),
        sa.Column("produto", sa.String(length=120), nullable=False),
        sa.Column("quantidade", sa.Integer(), nullable=False),
        sa.Column("preco_centavos", sa.Integer(), nullable=False),
        sa.ForeignKeyConstraint(["pedido_id"], ["pedidos.id"], ondelete="CASCADE"),
        sa.PrimaryKeyConstraint("id"),
    )


def downgrade() -> None:
    op.drop_table("itens_pedido")
    op.drop_index("ix_pedidos_cliente", table_name="pedidos")
    op.drop_table("pedidos")

    Aplicar, conferir e desfazer

    Terminal
    export API_DATABASE_URL="postgresql+psycopg://notes:notes@localhost:5432/notes_mig"
uv run alembic upgrade head
    Saída
    INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 0001, criar pedidos

    O PostgreSQL executa mudanças de esquema dentro de uma transação (transactional DDL), então uma migração que falha no meio é desfeita por inteiro. Esse é um dos motivos pelos quais eu o prefiro. Os comandos que você usa o tempo todo:

    Terminal
    uv run alembic current
uv run alembic check
uv run alembic downgrade base
uv run alembic upgrade head --sql
    Saída
    0001 (head)

    O check falha se os modelos mudaram sem uma migração correspondente (ele diz No new upgrade operations detected. quando está tudo sincronizado), e o --sql imprime o SQL sem executar nada, útil para revisão por quem administra o banco.

    Mudanças que mexem em dados

    Adicionar uma coluna NOT NULL a uma tabela que já tem linhas falharia, porque as linhas existentes não têm valor. O padrão seguro tem três passos na mesma migração: adicionar a coluna aceitando nulo, preencher as linhas existentes, e só então tornar obrigatória:

    Uma migração que preenche dados
    def upgrade() -> None:
    op.add_column("pedidos", sa.Column("canal", sa.String(length=20), nullable=True))
    op.execute("UPDATE pedidos SET canal = 'web' WHERE canal IS NULL")
    op.alter_column("pedidos", "canal", existing_type=sa.String(length=20), nullable=False)


def downgrade() -> None:
    op.drop_column("pedidos", "canal")

    Para implantar sem parar o sistema, o padrão se chama expandir e contrair: primeiro uma migração que só acrescenta (compatível com o código antigo), depois o deploy do código novo, e só numa migração posterior a remoção do que ficou obsoleto. Nunca renomeie ou apague uma coluna na mesma versão em que o código deixa de usá-la.

    Regras que eu não quebro

    Nunca edite uma migração que já foi aplicada em outro ambiente: crie uma nova. Mantenha uma única head (duas linhas de migração em paralelo precisam ser unificadas com alembic merge). O autogenerate não detecta renomeação de coluna: ele vê uma remoção e uma criação, o que apagaria os dados. Para renomear, escreva op.alter_column(..., new_column_name=...) à mão.

    No SQLite, alterar tabela é diferente

    O SQLite não consegue alterar colunas existentes. O Alembic oferece o modo batch (render_as_batch=True), que recria a tabela. Por isso eu rodo as migrações contra PostgreSQL, o mesmo banco da produção, e evito ter o SQLite como "banco de teste" das migrações.